feat(sql): named-query tag model — read-only SELECTs declared on the driver (lmxopcua#753)

SqlTagModel.Query (design §5.4) is now served end to end instead of refused.
The driver config declares `queries` (name -> sql, params, maxRows); a tag names
the query, the result `column` it reads and optionally a `rowKey` picking its row.

- Parser/deploy gate: query + column required, rowKey all-or-nothing; Inspect
  explains each refusal. Table is not read for this model.
- SqlReadOnlyStatement: single SELECT/WITH only, no data/schema-changing keyword
  outside literals/comments (refuses SELECT INTO, FOR UPDATE, DML CTEs). Shared by
  the factory and the AdminUI driver form (SqlNamedQueryRules.Describe), so the
  form refuses what the driver refuses. One bad entry fails the driver.
- Parameters are always bound; the factory refuses a declared parameter whose
  dialect marker the statement never uses.
- Reader: one execution per query per poll, inside a transaction that is always
  rolled back; result columns matched exact-then-unique-CI; no row -> BadNoData,
  ambiguous row / missing column / over maxRows -> BadConfigurationError.
- Catalog gate passes Query tags through (nothing is quoted into SQL); a
  Query-only driver skips the catalog load. A tag naming an undeclared query is
  skipped with the reason on /hosts.
- AdminUI: typed editor offers Query (query/column/rowKey); driver form edits
  `queries` as JSON and keeps the last valid set on a typo.

(lmxopcua#753)

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV

docs/drivers/Sql.md

@@ -37,7 +37,7 @@ who believes writes are enabled can. → #753
 
 ## Tag models
 
-A tag says how to find its value. Two models ship; a third is deferred.
+A tag says how to find its value. Three models ship: `KeyValue`, `WideRow` and the named-query `Query`.
 
 ### `KeyValue` — the EAV shape
 
@@ -109,11 +109,53 @@ canonical text form. SQL Server behaves the same way.
 authored `keyValue` refused the tag while the identically-bound `whereValue` was served — and the
 deploy gate reported the key as *missing* when it was right there in the blob.)
 
-### `Query` — deferred to P3 (→ #753)
+### `Query` — a named read-only query (#753)
 
-Named arbitrary-`SELECT`. **The v1 parser rejects it**, and that refusal is enforced end-to-end —
-parser, validator, planner and the AdminUI editor all decline it rather than accepting the value and
-ignoring it later.
+For a join, a view you cannot create, or a computed projection. The **driver config** declares the
+statement under `queries`; the **tag** names the query, the result column it reads, and optionally the
+row:
+
+```json
+{ "queries": {
+    "activeBatch": {
+      "sql": "SELECT line, speed, temp, ts FROM dbo.LineStatus WHERE active = @active",
+      "params": { "active": true },
+      "maxRows": 100
+    } } }
+```
+
+```json
+{ "model": "Query", "query": "activeBatch", "column": "speed",
+  "rowKey": { "column": "line", "value": "Line1" }, "timestampColumn": "ts", "type": "Float64" }
+```
+
+- **One execution per query per poll**, however many tags read it.
+- **`rowKey` picks the row** (matched as text against that result column, like `keyValue`). Without a
+  `rowKey` the query must return exactly one row. No matching row ⇒ `BadNoData`; two matching rows (or
+  two rows with no `rowKey`) ⇒ `BadConfigurationError` — never "last row wins".
+- **Result column names** (`column`, `rowKey.column`, `timestampColumn`) match exactly first, then by a
+  unique case-insensitive match, so Postgres' lower-cased and Oracle's upper-cased aliases still bind. A
+  column the query does not return ⇒ `BadConfigurationError` plus a rate-limited warning listing what it
+  did return.
+- **`maxRows`** (default 1000, at most 100 000) is enforced client-side: the reader stops one row past
+  it and the whole query publishes `BadConfigurationError`. A statement cannot be wrapped in a provider
+  row limit portably, so narrow the `WHERE` rather than relying on the cap.
+- **Parameters are always bound.** Keys are bare names (`active`; a leading `@`/`:` is tolerated);
+  values are JSON scalars or `null`. Write the provider's own marker in the SQL: `@active` on SqlServer,
+  Postgres and MySql, `:p_active` on Oracle, a positional `?` on Odbc (bound in authored order). The
+  factory refuses a declared parameter whose marker the statement never uses.
+- **Read-only, twice over.** `SqlReadOnlyStatement` refuses at construction (and in the AdminUI form)
+  anything but a single `SELECT`/`WITH` statement, and any of `INSERT UPDATE DELETE MERGE UPSERT DROP
+  ALTER CREATE TRUNCATE EXEC EXECUTE CALL GRANT REVOKE DENY INTO COPY LOCK RENAME` outside a literal or
+  comment — so `SELECT … INTO`, `FOR UPDATE` and a data-modifying CTE are refused too. Behind that guard
+  the statement runs **inside a transaction the driver always rolls back**, so a side effect the guard
+  cannot see (a function that writes) is never committed. A non-transactional side effect (an e-mail
+  procedure, a sequence) is outside both, so give the connecting principal `SELECT` only.
+- A tag naming a query the driver config does not declare is **skipped** at Initialize (it reads
+  `BadNodeIdUnknown`, and `/hosts` names the missing query). One unusable `queries` entry **fails the
+  driver** instead, because every tag that names it would otherwise go dark unexplained.
+- Query tags read no table, so they need **no catalog gate** — a driver serving only named queries skips
+  the catalog load entirely. Nothing from a Query tag is ever quoted into SQL.
 
 ## Optional per-tag fields
 
@@ -151,6 +193,7 @@ existed in the parser — nothing reads it.)
 | `maxConcurrentGroups` | `4` | Connection-pool guard: cap on group queries in flight. |
 | `nullIsBad` | `false` | `false` ⇒ a NULL cell publishes **Uncertain**; `true` ⇒ **Bad**. |
 | `allowWrites` | — | **Inert.** v1 has no write path; authoring `true` logs a warning. |
+| `queries` | — | Named read-only queries for `Query` tags — see [`Query`](#query--a-named-read-only-query-753). Edited as JSON in the driver form, which applies the factory's own rules. |
 
 ### Credentials never enter the deployment artifact
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlDriverConfigDto.cs

@@ -51,6 +51,13 @@ public sealed class SqlDriverConfigDto
     /// </summary>
     public bool? AllowWrites { get; init; }
 
+    /// <summary>
+    ///     Named read-only queries (design §5.1 / §5.4) a <see cref="SqlTagModel.Query"/> tag reads by name.
+    ///     Keys are matched ordinally. Each entry is validated by <see cref="SqlNamedQueryRules.Describe"/> at
+    ///     construction; one bad entry fails the driver rather than silently serving the rest.
+    /// </summary>
+    public Dictionary<string, SqlNamedQueryDto>? Queries { get; init; }
+
     /// <summary>
     ///     The authored raw tags the deploy artifact delivers for this driver instance — RawPath identity plus
     ///     the driver <c>TagConfig</c> blob, the same shape every other driver receives. The factory copies

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlEquipmentTagParser.cs

@@ -4,7 +4,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
 
 /// <summary>
-///     Pure mapper from an authored raw tag's <c>TagConfig</c> JSON (design §5.2 / §5.3) to a
+///     Pure mapper from an authored raw tag's <c>TagConfig</c> JSON (design §5.2 / §5.3 / §5.4) to a
 ///     <see cref="SqlTagDefinition"/>, mirroring <c>ModbusTagDefinitionFactory.FromTagConfig</c>. The blob is
 ///     recognised by a leading <c>{</c> plus a <c>"driver":"Sql"</c> marker or a <c>"model"</c> discriminator;
 ///     the <c>model</c> and <c>type</c> enums are read with
@@ -21,8 +21,9 @@ public static class SqlEquipmentTagParser
     ///     Maps an authored <c>TagConfig</c> blob to a typed definition keyed by <paramref name="rawPath"/>.
     ///     Returns <see langword="false"/> — never throws — for anything the driver must not serve: a
     ///     non-object / unparseable blob, a blob belonging to another driver, a typo'd <c>model</c> or
-    ///     <c>type</c> enum, a missing model-required field, or the P3-deferred
-    ///     <see cref="SqlTagModel.Query"/> model.
+    ///     <c>type</c> enum, or a missing model-required field. A <see cref="SqlTagModel.Query"/> tag needs
+    ///     <c>query</c> + <c>column</c> (and, when present, a complete <c>rowKey</c>); whether the named query
+    ///     exists is a driver-config fact the driver checks at Initialize.
     /// </summary>
     /// <param name="reference">The authored equipment-tag TagConfig JSON.</param>
     /// <param name="rawPath">The tag's RawPath — becomes the definition's identity (<c>Name</c>).</param>
@@ -53,9 +54,35 @@ public static class SqlEquipmentTagParser
                     ? type
                     : null;
 
+            var timestampColumn = ReadString(root, "timestampColumn");
+
+            if (model == SqlTagModel.Query)
+            {
+                // A named query reads no table of its own: the tag names the driver-config query, the result
+                // column it reads and — optionally — the row. A half-authored rowKey is refused rather than
+                // read as "no rowKey", which would silently demand a one-row result instead.
+                var queryName = ReadString(root, "query");
+                var column = ReadString(root, "column");
+                if (string.IsNullOrWhiteSpace(queryName) || string.IsNullOrWhiteSpace(column)) return false;
+                string? rowKeyColumn = null, rowKeyValue = null;
+                if (root.TryGetProperty("rowKey", out var rowKey))
+                {
+                    if (rowKey.ValueKind != JsonValueKind.Object) return false;
+                    rowKeyColumn = ReadString(rowKey, "column");
+                    rowKeyValue = ReadScalar(rowKey, "value");
+                    if (string.IsNullOrWhiteSpace(rowKeyColumn) || rowKeyValue is null) return false;
+                }
+
+                def = new SqlTagDefinition(
+                    Name: rawPath, Model: model, Table: string.Empty,
+                    TimestampColumn: timestampColumn, ColumnName: column,
+                    DeclaredType: declaredType, QueryName: queryName,
+                    RowKeyColumn: rowKeyColumn, RowKeyValue: rowKeyValue);
+                return true;
+            }
+
             var table = ReadString(root, "table");
             if (string.IsNullOrWhiteSpace(table)) return false;
-            var timestampColumn = ReadString(root, "timestampColumn");
 
             switch (model)
             {
@@ -100,8 +127,6 @@ public static class SqlEquipmentTagParser
                 }
 
                 default:
-                    // SqlTagModel.Query is design §5.4 / P3 — deferred. Reject rather than serve a model
-                    // the runtime cannot read.
                     return false;
             }
         }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlNamedQuery.cs

@@ -0,0 +1,298 @@
+using System.Text;
+using System.Text.Json;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
+
+/// <summary>
+///     One entry of a Sql driver config's <c>queries</c> map (design §5.1 / §5.4) — a named, read-only
+///     statement a <see cref="SqlTagModel.Query"/> tag reads by name.
+///     <para><b>The SQL is configuration, not tag input.</b> It is authored on the driver by a
+///     ConfigEditor-privileged operator and travels in the deployment artifact. Runtime values still never
+///     reach the text: every <see cref="Params"/> entry is bound as a <c>DbParameter</c>.</para>
+/// </summary>
+public sealed class SqlNamedQueryDto
+{
+    /// <summary>
+    ///     The statement — exactly one <c>SELECT</c> or <c>WITH … SELECT</c>. Parameter markers are written in
+    ///     the provider's own spelling for each <see cref="Params"/> key (<c>@name</c> on SqlServer, Postgres
+    ///     and MySql; <c>:p_name</c> on Oracle; positional <c>?</c> on Odbc, in authored order). Checked by
+    ///     <see cref="SqlReadOnlyStatement.Check"/>.
+    /// </summary>
+    public string? Sql { get; init; }
+
+    /// <summary>
+    ///     Bound parameter values, keyed by the bare parameter name (a leading <c>@</c> or <c>:</c> is
+    ///     tolerated and stripped). Values must be JSON scalars or <c>null</c>.
+    /// </summary>
+    public Dictionary<string, JsonElement>? Params { get; init; }
+
+    /// <summary>
+    ///     The most rows one poll reads before the whole query is refused as ambiguous
+    ///     (<c>BadConfigurationError</c>). Absent ⇒ <see cref="SqlNamedQueryRules.DefaultMaxRows"/>.
+    /// </summary>
+    public int? MaxRows { get; init; }
+}
+
+/// <summary>Limits and name rules shared by the factory's named-query parse and the AdminUI form.</summary>
+public static class SqlNamedQueryRules
+{
+    /// <summary>The default per-poll row ceiling of a named query.</summary>
+    public const int DefaultMaxRows = 1000;
+
+    /// <summary>The highest <c>maxRows</c> that may be authored.</summary>
+    public const int MaxMaxRows = 100_000;
+
+    /// <summary>
+    ///     Normalises an authored parameter key to its bare base name — strips one leading <c>@</c> or
+    ///     <c>:</c> — and returns <see langword="null"/> when the rest is not a plain identifier
+    ///     (<c>[A-Za-z_][A-Za-z0-9_]*</c>, at most 30 characters, which is Oracle's bind-name ceiling). The
+    ///     base name feeds the dialect's own marker and bind-name spelling, so it must not carry punctuation.
+    /// </summary>
+    /// <param name="key">The authored key.</param>
+    /// <returns>The bare base name, or null when unusable.</returns>
+    public static string? NormaliseParameterName(string? key)
+    {
+        if (string.IsNullOrEmpty(key)) return null;
+        var name = key[0] is '@' or ':' ? key[1..] : key;
+        if (name.Length is 0 or > 30) return null;
+        if (!(char.IsAsciiLetter(name[0]) || name[0] == '_')) return null;
+        foreach (var ch in name)
+            if (!(char.IsAsciiLetterOrDigit(ch) || ch == '_')) return null;
+        return name;
+    }
+
+    /// <summary>
+    ///     Validates one named query's provider-independent shape — name, statement, parameter names and
+    ///     values, row ceiling — and returns the first problem, or <see langword="null"/> when it is usable.
+    ///     The runtime factory and the AdminUI driver form both call this, so what the form accepts is what
+    ///     the driver constructs. (The factory additionally checks each parameter's marker appears in the
+    ///     statement, which needs the provider's dialect.)
+    /// </summary>
+    /// <param name="name">The <c>queries</c> key.</param>
+    /// <param name="query">The entry.</param>
+    /// <returns>An operator-facing error, or null.</returns>
+    public static string? Describe(string name, SqlNamedQueryDto? query)
+    {
+        if (string.IsNullOrWhiteSpace(name)) return "a named query has a blank name";
+        if (query is null) return $"named query '{name}' is null";
+
+        var statement = SqlReadOnlyStatement.Check(query.Sql, out _);
+        if (statement is not null) return $"named query '{name}': {statement}";
+
+        if (query.MaxRows is { } maxRows && (maxRows < 1 || maxRows > MaxMaxRows))
+            return $"named query '{name}' has maxRows {maxRows}; it must be between 1 and {MaxMaxRows}";
+
+        if (query.Params is { } parameters)
+        {
+            var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
+            foreach (var (key, value) in parameters)
+            {
+                var bare = NormaliseParameterName(key);
+                if (bare is null)
+                    return $"named query '{name}' has parameter '{key}', which is not a plain identifier " +
+                           "(letters, digits and '_', starting with a letter or '_', at most 30 characters)";
+                if (!seen.Add(bare))
+                    return $"named query '{name}' declares parameter '{bare}' more than once";
+                if (value.ValueKind is not (JsonValueKind.String or JsonValueKind.Number or JsonValueKind.True
+                    or JsonValueKind.False or JsonValueKind.Null))
+                    return $"named query '{name}' parameter '{bare}' must be a string, number, boolean or null";
+            }
+        }
+
+        return null;
+    }
+}
+
+/// <summary>
+///     The authoring-time read-only guard for a named query's SQL (design §5.4). It is a <b>guard, not the
+///     boundary</b>: the driver also runs every named query inside a transaction it always rolls back, and the
+///     connecting principal should hold <c>SELECT</c> only. The guard exists so an operator learns at
+///     construction — not from a database audit — that a statement is not a plain read.
+///     <para>Rules, applied to the text <em>outside</em> string literals, quoted identifiers and comments:
+///     the statement begins with <c>SELECT</c> or <c>WITH</c>; it is a single statement (one trailing
+///     <c>;</c> is tolerated and removed); and none of the data- or schema-changing keywords appears anywhere
+///     — which also refuses <c>SELECT … INTO</c>, <c>FOR UPDATE</c> and a data-modifying CTE.</para>
+/// </summary>
+public static class SqlReadOnlyStatement
+{
+    /// <summary>
+    ///     Keywords that may not appear outside literals/comments in a named query. Statement-only keywords
+    ///     (<c>COMMIT</c>, <c>BEGIN</c>, …) are not listed: the single-statement + <c>SELECT</c>-first rules
+    ///     already exclude them, and listing them would refuse a column that merely shares the name. A column
+    ///     named like a listed keyword can still be read by quoting it.
+    /// </summary>
+    private static readonly HashSet<string> Forbidden = new(StringComparer.OrdinalIgnoreCase)
+    {
+        "INSERT", "UPDATE", "DELETE", "MERGE", "UPSERT", "DROP", "ALTER", "CREATE", "TRUNCATE",
+        "EXEC", "EXECUTE", "CALL", "GRANT", "REVOKE", "DENY", "INTO", "COPY", "LOCK", "RENAME",
+    };
+
+    /// <summary>
+    ///     Checks <paramref name="sql"/> and returns the problem, or <see langword="null"/> when it is a single
+    ///     read-only statement. On success <paramref name="normalised"/> is the statement with surrounding
+    ///     whitespace and one trailing <c>;</c> removed (ODP.NET refuses a trailing terminator).
+    /// </summary>
+    /// <param name="sql">The authored statement.</param>
+    /// <param name="normalised">The statement to execute, when valid.</param>
+    /// <returns>An operator-facing error, or null.</returns>
+    public static string? Check(string? sql, out string normalised)
+    {
+        normalised = string.Empty;
+        if (string.IsNullOrWhiteSpace(sql)) return "the statement is empty";
+
+        var words = new List<string>();
+        var terminatorAt = -1;
+        var i = 0;
+        while (i < sql.Length)
+        {
+            var ch = sql[i];
+            var next = i + 1 < sql.Length ? sql[i + 1] : '\0';
+
+            if (ch == '-' && next == '-')
+            {
+                while (i < sql.Length && sql[i] != '\n') i++;
+                continue;
+            }
+
+            if (ch == '/' && next == '*')
+            {
+                var end = sql.IndexOf("*/", i + 2, StringComparison.Ordinal);
+                if (end < 0) return "the statement has an unterminated /* comment";
+                i = end + 2;
+                continue;
+            }
+
+            if (ch is '\'' or '"' or '`' or '[')
+            {
+                var close = ch == '[' ? ']' : ch;
+                var j = i + 1;
+                while (true)
+                {
+                    if (j >= sql.Length) return "the statement has an unterminated quoted literal or identifier";
+                    if (sql[j] == close)
+                    {
+                        // A doubled closing character is an escaped one inside the literal.
+                        if (j + 1 < sql.Length && sql[j + 1] == close && close != ']') { j += 2; continue; }
+                        break;
+                    }
+
+                    j++;
+                }
+
+                if (terminatorAt >= 0) return "the statement carries more than one statement (text after ';')";
+                i = j + 1;
+                continue;
+            }
+
+            if (ch == ';')
+            {
+                if (terminatorAt >= 0) return "the statement carries more than one statement (';' repeated)";
+                terminatorAt = i;
+                i++;
+                continue;
+            }
+
+            if (char.IsLetter(ch) || ch == '_')
+            {
+                var start = i;
+                while (i < sql.Length && (char.IsLetterOrDigit(sql[i]) || sql[i] is '_' or '$' or '#')) i++;
+                if (terminatorAt >= 0) return "the statement carries more than one statement (text after ';')";
+                words.Add(sql[start..i]);
+                continue;
+            }
+
+            if (!char.IsWhiteSpace(ch) && terminatorAt >= 0)
+                return "the statement carries more than one statement (text after ';')";
+            i++;
+        }
+
+        if (words.Count == 0) return "the statement is empty";
+        if (!words[0].Equals("SELECT", StringComparison.OrdinalIgnoreCase)
+            && !words[0].Equals("WITH", StringComparison.OrdinalIgnoreCase))
+            return "a named query must be a single SELECT (or WITH … SELECT) statement";
+
+        foreach (var word in words)
+        {
+            if (Forbidden.Contains(word))
+                return $"a named query must be read-only, and '{word.ToUpperInvariant()}' is not allowed outside a " +
+                       "string literal or comment";
+        }
+
+        var body = terminatorAt >= 0 ? sql[..terminatorAt] : sql;
+        normalised = body.Trim();
+        return null;
+    }
+
+    /// <summary>
+    ///     True when <paramref name="marker"/> appears in <paramref name="sql"/> as a parameter token — outside
+    ///     literals and comments, and not as the prefix of a longer name (<c>@line</c> inside <c>@lineId</c>).
+    /// </summary>
+    /// <param name="sql">The statement.</param>
+    /// <param name="marker">The provider marker, e.g. <c>@line</c> or <c>:p_line</c>.</param>
+    /// <returns>True when the marker is used.</returns>
+    public static bool UsesMarker(string sql, string marker)
+    {
+        ArgumentNullException.ThrowIfNull(sql);
+        ArgumentException.ThrowIfNullOrEmpty(marker);
+
+        var code = StripLiteralsAndComments(sql);
+        var from = 0;
+        while (true)
+        {
+            var at = code.IndexOf(marker, from, StringComparison.OrdinalIgnoreCase);
+            if (at < 0) return false;
+            var after = at + marker.Length;
+            if (after >= code.Length || !(char.IsLetterOrDigit(code[after]) || code[after] == '_')) return true;
+            from = after;
+        }
+    }
+
+    /// <summary>Replaces literal, quoted-identifier and comment content with spaces, keeping offsets.</summary>
+    private static string StripLiteralsAndComments(string sql)
+    {
+        var builder = new StringBuilder(sql);
+        var i = 0;
+        while (i < sql.Length)
+        {
+            var ch = sql[i];
+            var next = i + 1 < sql.Length ? sql[i + 1] : '\0';
+            int end;
+            if (ch == '-' && next == '-')
+            {
+                end = sql.IndexOf('\n', i);
+                end = end < 0 ? sql.Length : end;
+            }
+            else if (ch == '/' && next == '*')
+            {
+                end = sql.IndexOf("*/", i + 2, StringComparison.Ordinal);
+                end = end < 0 ? sql.Length : end + 2;
+            }
+            else if (ch is '\'' or '"' or '`' or '[')
+            {
+                var close = ch == '[' ? ']' : ch;
+                end = i + 1;
+                while (end < sql.Length)
+                {
+                    if (sql[end] == close)
+                    {
+                        if (end + 1 < sql.Length && sql[end + 1] == close && close != ']') { end += 2; continue; }
+                        end++;
+                        break;
+                    }
+
+                    end++;
+                }
+            }
+            else
+            {
+                i++;
+                continue;
+            }
+
+            for (var k = i; k < end; k++) builder[k] = ' ';
+            i = end;
+        }
+
+        return builder.ToString();
+    }
+}

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlProvider.cs

@@ -23,7 +23,7 @@ public enum SqlProvider
     Oracle,
 }
 
-/// <summary>Tag→value mapping model. v1 ships KeyValue + WideRow; Query is deferred (design §5.4 / P3).</summary>
+/// <summary>Tag→value mapping model: KeyValue, WideRow, and the named-query model Query (design §5.4).</summary>
 public enum SqlTagModel
 {
     /// <summary>
@@ -38,6 +38,11 @@ public enum SqlTagModel
     /// </summary>
     WideRow,
 
-    /// <summary>Named arbitrary-SELECT query (design §5.4). <b>Deferred to P3 — not accepted by v1's parser.</b></summary>
+    /// <summary>
+    ///     Named read-only query (design §5.4): the driver config declares <c>queries</c>, each a single
+    ///     <c>SELECT</c>/<c>WITH</c> statement with bound parameters; the tag names a <c>query</c>, the result
+    ///     <c>column</c> it reads, and optionally a <c>rowKey</c> picking its row. The SQL is configuration
+    ///     authored by a ConfigEditor, never tag input, and runs inside a rolled-back transaction.
+    /// </summary>
     Query,
 }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinition.cs

@@ -9,7 +9,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
 ///     into SQL.</b> Value-bearing fields (<see cref="KeyValue"/>, <see cref="RowSelectorValue"/>) are bound
 ///     as <c>DbParameter</c>s by the reader; identifier-bearing fields (<see cref="Table"/>,
 ///     <see cref="KeyColumn"/>, <see cref="ValueColumn"/>, <see cref="TimestampColumn"/>,
-///     <see cref="ColumnName"/>, <see cref="RowSelectorColumn"/>,
+///     <see cref="ColumnName"/> (for <see cref="SqlTagModel.Query"/>: the result-set column), <see cref="RowSelectorColumn"/>,
 ///     <see cref="RowSelectorTopByTimestamp"/>) must be validated against <c>INFORMATION_SCHEMA</c> and
 ///     dialect-quoted by the reader before they reach a command text. Sanitising here would corrupt
 ///     legitimate values, so the parser deliberately does not.</para>
@@ -18,8 +18,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
 ///     The tag's identity — its <b>RawPath</b>, exactly as handed to the parser. The driver's forward
 ///     router keys published values on this, so it must never be derived from the blob's contents.
 /// </param>
-/// <param name="Model">Which tag→value mapping this tag uses. v1 accepts KeyValue and WideRow only.</param>
-/// <param name="Table">The table or view to read (e.g. <c>dbo.TagValues</c>).</param>
+/// <param name="Model">Which tag→value mapping this tag uses: KeyValue, WideRow or Query.</param>
+/// <param name="Table">The table or view to read (e.g. <c>dbo.TagValues</c>). Empty for <see cref="SqlTagModel.Query"/>, which reads a named query instead.</param>
 /// <param name="KeyColumn"><see cref="SqlTagModel.KeyValue"/>: the column matched against <see cref="KeyValue"/>.</param>
 /// <param name="KeyValue"><see cref="SqlTagModel.KeyValue"/>: this tag's key — <b>bound</b>, never interpolated.</param>
 /// <param name="ValueColumn"><see cref="SqlTagModel.KeyValue"/>: the column the value is read from.</param>
@@ -28,6 +28,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
 /// <param name="RowSelectorColumn"><see cref="SqlTagModel.WideRow"/>: the <c>whereColumn</c> that picks the row.</param>
 /// <param name="RowSelectorValue"><see cref="SqlTagModel.WideRow"/>: the <c>whereValue</c> — <b>bound</b>, never interpolated.</param>
 /// <param name="RowSelectorTopByTimestamp"><see cref="SqlTagModel.WideRow"/>: newest-row selector — the timestamp column to order by, when no where-pair is authored.</param>
+/// <param name="QueryName"><see cref="SqlTagModel.Query"/>: the driver-config <c>queries</c> entry this tag reads.</param>
+/// <param name="RowKeyColumn">
+///     <see cref="SqlTagModel.Query"/>: the result-set column that picks this tag's row. Null ⇒ the query must
+///     return exactly one row. Matched against the result set, never quoted into SQL.
+/// </param>
+/// <param name="RowKeyValue"><see cref="SqlTagModel.Query"/>: the <see cref="RowKeyColumn"/> value (matched as text).</param>
 /// <param name="DeclaredType">
 ///     Optional explicit type override from the blob's <c>type</c> field. <see langword="null"/> ⇒ the
 ///     driver infers the type from the result set's column metadata.
@@ -44,4 +50,7 @@ public sealed record SqlTagDefinition(
     string? RowSelectorColumn = null,
     string? RowSelectorValue = null,
     string? RowSelectorTopByTimestamp = null,
-    DriverDataType? DeclaredType = null);
+    DriverDataType? DeclaredType = null,
+    string? QueryName = null,
+    string? RowKeyColumn = null,
+    string? RowKeyValue = null);

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinitionFactory.cs

@@ -37,8 +37,8 @@ public static class SqlTagDefinitionFactory
     ///     Deploy-time inspection of an equipment-tag <c>TagConfig</c> blob. Returns human-readable warnings
     ///     for a blob <see cref="SqlEquipmentTagParser.TryParse"/> would reject — an unparseable or
     ///     non-object document, a blob the driver does not recognise as a Sql one, a present-but-invalid
-    ///     <c>model</c> / <c>type</c> enum, the P3-deferred <c>Query</c> model, a missing <c>table</c>, or a
-    ///     missing model-required field. Empty when the parser accepts the blob, when the blob is blank, or
+    ///     <c>model</c> / <c>type</c> enum, a missing <c>table</c> (KeyValue/WideRow), or a missing
+    ///     model-required field (a <c>Query</c> tag's <c>query</c>/<c>column</c>/<c>rowKey</c>). Empty when the parser accepts the blob, when the blob is blank, or
     ///     when it is not a JSON object at all (the sibling factories' "not an equipment-tag TagConfig"
     ///     convention). Never throws.
     /// </summary>
@@ -113,9 +113,20 @@ public static class SqlTagDefinitionFactory
         var model = TagConfigJson.ReadEnumOrDefault(root, "model", SqlTagModel.KeyValue);
         if (model == SqlTagModel.Query)
         {
-            warnings.Add(
-                "Sql tag model 'Query' is deferred (design §5.4 / P3) and is not served — the driver will "
-                + "skip this tag.");
+            if (string.IsNullOrWhiteSpace(ReadString(root, "query")))
+                warnings.Add("Sql Query tag has no 'query' — name the driver-config named query it reads.");
+            if (string.IsNullOrWhiteSpace(ReadString(root, "column")))
+                warnings.Add("Sql Query tag has no 'column' — name the result-set column it reads.");
+            if (root.TryGetProperty("rowKey", out var rowKey)
+                && (rowKey.ValueKind != JsonValueKind.Object
+                    || string.IsNullOrWhiteSpace(ReadString(rowKey, "column"))
+                    || !HasScalar(rowKey, "value")))
+            {
+                warnings.Add(
+                    "Sql Query tag has an incomplete 'rowKey' — it needs both a 'column' and a scalar 'value' "
+                    + "(omit 'rowKey' entirely when the query returns exactly one row).");
+            }
+
             return;
         }
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalog.cs

@@ -40,6 +40,15 @@ public sealed class SqlCatalog
         _columnsByTable = columnsByTable;
     }
 
+    /// <summary>
+    ///     A catalog that resolves nothing — used when no authored tag needs one (a driver serving only named
+    ///     queries), so Initialize does not issue catalog queries whose answer nothing reads.
+    /// </summary>
+    internal static SqlCatalog Empty { get; } = new(
+        string.Empty, [],
+        new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal),
+        new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal));
+
     /// <summary>The schema an unqualified authored object name is resolved in.</summary>
     public string DefaultSchema { get; }
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogGate.cs

@@ -77,6 +77,16 @@ public static class SqlCatalogGate
         foreach (var definition in definitions)
         {
             ArgumentNullException.ThrowIfNull(definition);
+
+            // A named-query tag quotes no identifier into SQL: its statement is driver configuration (checked
+            // read-only at construction) and its column/rowKey/timestamp names are matched against the result
+            // set by the reader. There is nothing here to allow-list, so it passes through unchanged.
+            if (definition.Model == SqlTagModel.Query)
+            {
+                accepted.Add(definition);
+                continue;
+            }
+
             if (TryCanonicalize(definition, catalog, dialect, out var canonical, out var rejection))
                 accepted.Add(canonical!);
             else

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs

@@ -290,7 +290,8 @@ public sealed class SqlDriver
             maxConcurrentGroups: options.MaxConcurrentGroups,
             nullIsBad: options.NullIsBad,
             resolve: rawPath => _resolver.TryResolve(rawPath, out var definition) ? definition : null,
-            logger: _logger);
+            logger: _logger,
+            queries: options.Queries);
         previous?.Dispose();
     }
 
@@ -966,7 +967,25 @@ public sealed class SqlDriver
             try
             {
                 if (SqlEquipmentTagParser.TryParse(entry.TagConfig, entry.RawPath, out var definition))
+                {
+                    if (definition.Model == SqlTagModel.Query
+                        && !_options.Queries.ContainsKey(definition.QueryName!))
+                    {
+                        // The blob is well-formed but names a query this driver's config does not declare — a
+                        // fact only the driver can see (the deploy gate inspects one tag blob at a time). Skip
+                        // it like any unmappable tag, so it reads BadNodeIdUnknown and /hosts says why.
+                        var missing =
+                            $"reads named query '{definition.QueryName}', which this driver's config does not " +
+                            "declare under 'queries'";
+                        blobSkips.Add(new SkippedTag(entry.RawPath, missing));
+                        _logger.LogWarning(
+                            "Sql tag skipped. Driver={DriverInstanceId} RawPath={RawPath} Reason={Reason}",
+                            _driverInstanceId, entry.RawPath, missing);
+                        continue;
+                    }
+
                     table[entry.RawPath] = definition;
+                }
                 else
                 {
                     // The reason is the DEPLOY GATE's own explanation (R06:S-4) — SqlTagDefinitionFactory.Inspect
@@ -1038,17 +1057,22 @@ public sealed class SqlDriver
         var authored = Tags;
         if (authored.Count == 0) return;
 
+        // Named-query tags read no table of their own and quote no identifier, so they need no catalog —
+        // SqlCatalogGate passes them through. A driver serving ONLY named queries therefore skips the load.
         var tables = authored.Values
+            .Where(definition => definition.Model != SqlTagModel.Query)
             .Select(definition => definition.Table)
             .Where(table => !string.IsNullOrWhiteSpace(table))
             .Distinct(StringComparer.OrdinalIgnoreCase)
             .ToArray();
 
-        var catalog = await RunBoundedAsync(
-            token => LoadCatalogAsync(tables, token),
-            _options.OperationTimeout,
-            "the catalog queries did not return",
-            cancellationToken).ConfigureAwait(false);
+        var catalog = tables.Length == 0
+            ? SqlCatalog.Empty
+            : await RunBoundedAsync(
+                token => LoadCatalogAsync(tables, token),
+                _options.OperationTimeout,
+                "the catalog queries did not return",
+                cancellationToken).ConfigureAwait(false);
 
         var result = SqlCatalogGate.Apply(authored.Values, catalog, _dialect);
 

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriverFactoryExtensions.cs

@@ -115,7 +115,7 @@ public static class SqlDriverFactoryExtensions
         // Resolved BEFORE the knob validation in CreateInstance, so that every subsequent throw is a path
         // that HAS the secret in hand — which is exactly where a careless interpolation would leak it.
         var connectionString = SqlConnectionStringResolver.Resolve(dto.ConnectionStringRef!);
-        return new SqlDriverBinding(BuildOptions(dto, driverInstanceId), dialect, connectionString, dto);
+        return new SqlDriverBinding(BuildOptions(dto, driverInstanceId, dialect), dialect, connectionString, dto);
     }
 
     /// <summary>
@@ -237,9 +237,16 @@ public static class SqlDriverFactoryExtensions
     /// </summary>
     /// <param name="dto">The deserialised config blob.</param>
     /// <param name="driverInstanceId">Named in every rejection message, so a fleet-wide log identifies the row.</param>
+    /// <param name="dialect">
+    ///     The provider seam, for the named-query marker check; null skips that one check (unit tests of the
+    ///     other knobs).
+    /// </param>
     /// <returns>The typed options the driver is constructed with.</returns>
-    /// <exception cref="InvalidOperationException">A knob is out of range, or the two deadlines are inverted.</exception>
-    internal static SqlDriverOptions BuildOptions(SqlDriverConfigDto dto, string driverInstanceId)
+    /// <exception cref="InvalidOperationException">
+    ///     A knob is out of range, the two deadlines are inverted, or a named query is unusable.
+    /// </exception>
+    internal static SqlDriverOptions BuildOptions(
+        SqlDriverConfigDto dto, string driverInstanceId, ISqlDialect? dialect = null)
     {
         ArgumentNullException.ThrowIfNull(dto);
 
@@ -280,9 +287,71 @@ public static class SqlDriverFactoryExtensions
             CommandTimeout = commandTimeout,
             MaxConcurrentGroups = maxConcurrentGroups,
             NullIsBad = dto.NullIsBad ?? false,
+            Queries = BuildQueries(dto, driverInstanceId, dialect),
         };
     }
 
+    /// <summary>
+    ///     Validates and types the config's named queries (design §5.4). <b>One unusable entry fails the
+    ///     driver</b>: unlike a tag, a query is shared by every tag that names it, and a driver that silently
+    ///     dropped it would turn all of them into an unexplained <c>BadNodeIdUnknown</c>.
+    ///     <para>The shape rules are <see cref="SqlNamedQueryRules.Describe"/> — the same call the AdminUI driver
+    ///     form makes — so the form refuses what this refuses. The one rule added here needs the dialect: every
+    ///     declared parameter's marker (<see cref="ISqlDialect.ParameterMarker"/>) must appear in the statement.
+    ///     A provider that binds positionally (<c>?</c>) is exempt, since there is no name to look for.</para>
+    /// </summary>
+    private static IReadOnlyDictionary<string, SqlNamedQuery> BuildQueries(
+        SqlDriverConfigDto dto, string driverInstanceId, ISqlDialect? dialect)
+    {
+        var queries = new Dictionary<string, SqlNamedQuery>(StringComparer.Ordinal);
+        if (dto.Queries is not { Count: > 0 } authored) return queries;
+
+        foreach (var (name, query) in authored)
+        {
+            if (SqlNamedQueryRules.Describe(name, query) is { } problem)
+                throw new InvalidOperationException($"Sql driver config for '{driverInstanceId}': {problem}.");
+
+            SqlReadOnlyStatement.Check(query.Sql, out var sql);
+            var names = new List<string>();
+            var values = new List<object?>();
+            foreach (var (key, value) in query.Params ?? [])
+            {
+                var bare = SqlNamedQueryRules.NormaliseParameterName(key)!;
+                if (dialect is not null)
+                {
+                    var marker = dialect.ParameterMarker(bare);
+                    if (marker != "?" && !SqlReadOnlyStatement.UsesMarker(sql, marker))
+                    {
+                        throw new InvalidOperationException(
+                            $"Sql driver config for '{driverInstanceId}': named query '{name}' declares parameter " +
+                            $"'{bare}', but its statement never uses the {dialect.Provider} marker '{marker}'.");
+                    }
+                }
+
+                names.Add(bare);
+                values.Add(ToParameterValue(value));
+            }
+
+            queries[name] = new SqlNamedQuery(
+                name, sql, names, values, query.MaxRows ?? SqlNamedQueryRules.DefaultMaxRows);
+        }
+
+        return queries;
+    }
+
+    /// <summary>
+    ///     Converts a validated JSON scalar to the CLR value bound for it: text, a 64-bit integer when the number
+    ///     is integral, otherwise a double, a boolean, or null (bound as <c>DBNull</c>).
+    /// </summary>
+    private static object? ToParameterValue(JsonElement value) => value.ValueKind switch
+    {
+        JsonValueKind.String => value.GetString(),
+        JsonValueKind.Number => value.TryGetInt64(out var integral) ? integral : value.GetDouble(),
+        JsonValueKind.True => true,
+        JsonValueKind.False => false,
+        _ => null,
+    };
+
     /// <summary>Rejects a non-positive duration, naming the authored field.</summary>
     private static void RequirePositive(TimeSpan value, string field, string driverInstanceId)
     {

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriverOptions.cs

@@ -53,4 +53,27 @@ public sealed class SqlDriverOptions
     ///     It never governs an absent row, which is always <see cref="SqlStatusCodes.BadNoData"/>.
     /// </summary>
     public bool NullIsBad { get; init; }
+
+    /// <summary>
+    ///     The validated named queries (design §5.4) a <see cref="SqlTagModel.Query"/> tag reads by name, keyed
+    ///     ordinally. Empty when the config declares none.
+    /// </summary>
+    public IReadOnlyDictionary<string, SqlNamedQuery> Queries { get; init; } =
+        new Dictionary<string, SqlNamedQuery>(StringComparer.Ordinal);
 }
+
+/// <summary>
+///     One validated named query — what <see cref="SqlNamedQueryDto"/> becomes once the factory has checked it
+///     (<see cref="SqlNamedQueryRules.Describe"/> plus the dialect's marker check).
+/// </summary>
+/// <param name="Name">The <c>queries</c> key tags reference.</param>
+/// <param name="Sql">The normalised single read-only statement (no trailing terminator).</param>
+/// <param name="ParameterNames">The bare parameter base names, in authored order.</param>
+/// <param name="ParameterValues">The values bound under <paramref name="ParameterNames"/>, aligned index-for-index.</param>
+/// <param name="MaxRows">The per-poll row ceiling; one row more refuses the whole query.</param>
+public sealed record SqlNamedQuery(
+    string Name,
+    string Sql,
+    IReadOnlyList<string> ParameterNames,
+    IReadOnlyList<object?> ParameterValues,
+    int MaxRows);

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlGroupPlanner.cs

@@ -56,12 +56,18 @@ public static class SqlGroupPlanner
     ///     <see cref="ISqlDialect.QuoteIdentifier"/>). <see cref="SqlEquipmentTagParser"/> rejects all of these
     ///     up front, so reaching one here means a caller built a definition by hand with a broken invariant.
     /// </exception>
-    /// <exception cref="NotSupportedException">
-    ///     A tag carries <see cref="SqlTagModel.Query"/> (design §5.4, deferred to P3).
-    ///     <b>Deliberately loud rather than skipped:</b> silently dropping a tag would leave an authored node
-    ///     permanently unfed with nothing in the log, and the parser makes this branch unreachable in practice.
-    /// </exception>
-    public static IReadOnlyList<SqlQueryPlan> Plan(IEnumerable<SqlTagDefinition> tags, ISqlDialect dialect)
+    /// <exception cref="NotSupportedException">A tag carries a model this planner does not know.</exception>
+    /// <param name="queries">
+    ///     The driver's validated named queries, which a <see cref="SqlTagModel.Query"/> tag needs: every tag
+    ///     naming one query folds into ONE plan running that query's own statement (design §5.4). Null when the
+    ///     driver declares none; a Query tag naming a query this does not hold throws
+    ///     <see cref="ArgumentException"/> (the driver drops such tags at Initialize, so that is a broken caller
+    ///     invariant).
+    /// </param>
+    public static IReadOnlyList<SqlQueryPlan> Plan(
+        IEnumerable<SqlTagDefinition> tags,
+        ISqlDialect dialect,
+        IReadOnlyDictionary<string, SqlNamedQuery>? queries = null)
     {
         ArgumentNullException.ThrowIfNull(tags);
         ArgumentNullException.ThrowIfNull(dialect);
@@ -97,6 +103,9 @@ public static class SqlGroupPlanner
                 case SqlTagModel.WideRow:
                     plans.Add(BuildWideRowPlan(key, members, dialect));
                     break;
+                case SqlTagModel.Query:
+                    plans.Add(BuildQueryPlan(key, members, dialect, queries));
+                    break;
                 default:
                     throw UnsupportedModel(members[0]);
             }
@@ -106,8 +115,8 @@ public static class SqlGroupPlanner
     }
 
     /// <summary>
-    ///     Composes the equality key that decides which tags share a query. Both models produce a 5-tuple whose
-    ///     first element is the model, so a key-value key can never compare equal to a wide-row key.
+    ///     Composes the equality key that decides which tags share a query. Every model produces a 5-tuple whose
+    ///     first element is the model, so keys of different models can never compare equal.
     ///     <list type="bullet">
     ///         <item>
     ///             <see cref="SqlTagModel.KeyValue"/> → <c>(model, table, keyColumn, valueColumn, timestampColumn)</c>.
@@ -131,9 +140,49 @@ public static class SqlGroupPlanner
         SqlTagModel.KeyValue => (tag.Model, tag.Table, tag.KeyColumn, tag.ValueColumn, tag.TimestampColumn),
         SqlTagModel.WideRow => (tag.Model, tag.Table, tag.RowSelectorColumn, tag.RowSelectorValue,
             tag.RowSelectorTopByTimestamp),
+        // One execution per named query, whatever columns or rows its members pick out of the result.
+        SqlTagModel.Query => (tag.Model, tag.QueryName, (string?)null, (string?)null, (string?)null),
         _ => throw UnsupportedModel(tag),
     };
 
+    /// <summary>
+    ///     Emits one plan for a named query: the query's own (already-validated, read-only) statement, with
+    ///     each declared parameter spelled by the dialect's marker and bound under its bind name.
+    ///     <para><b>No identifier from a tag reaches this text.</b> A Query tag's <c>column</c>,
+    ///     <c>rowKey.column</c> and <c>timestampColumn</c> are matched against the result set's column names
+    ///     by the reader; they are never quoted into SQL, so they need no catalog gate.</para>
+    ///     <para><b>The row cap is client-side only.</b> An arbitrary statement cannot be wrapped in a
+    ///     provider row limit portably (a T-SQL CTE cannot sit in a derived table), so
+    ///     <see cref="SqlQueryPlan.RowCap"/> is the query's <c>maxRows</c> and the reader stops reading one row
+    ///     past it.</para>
+    /// </summary>
+    private static SqlQueryPlan BuildQueryPlan(
+        object groupKey,
+        List<SqlTagDefinition> members,
+        ISqlDialect dialect,
+        IReadOnlyDictionary<string, SqlNamedQuery>? queries)
+    {
+        var first = members[0];
+        var name = first.QueryName;
+        if (string.IsNullOrWhiteSpace(name)) throw MissingField(nameof(SqlTagDefinition.QueryName), first);
+        if (queries is null || !queries.TryGetValue(name, out var query))
+            throw new ArgumentException(
+                $"Sql tag '{first.Name}' reads named query '{name}', which the driver config does not define.",
+                nameof(queries));
+
+        var markers = new List<string>(query.ParameterNames.Count);
+        var bindNames = new List<string>(query.ParameterNames.Count);
+        foreach (var parameter in query.ParameterNames)
+        {
+            markers.Add(dialect.ParameterMarker(parameter));
+            bindNames.Add(dialect.ParameterBindName(parameter));
+        }
+
+        return new SqlQueryPlan(
+            SqlTagModel.Query, groupKey, query.Sql, markers, bindNames, query.ParameterValues, members,
+            SelectedColumns: [], RowCap: query.MaxRows);
+    }
+
     /// <summary>
     ///     Emits one key-value plan per <b>chunk</b> of the group's distinct keys, sized by
     ///     <see cref="ISqlDialect.MaxBindParameters"/>. A group at or under the ceiling — every ordinary group
@@ -346,6 +395,5 @@ public static class SqlGroupPlanner
         => new($"Sql tag '{tag.Name}' ({tag.Model}) is missing the required '{field}'.");
 
     private static NotSupportedException UnsupportedModel(SqlTagDefinition tag)
-        => new($"Sql tag '{tag.Name}' uses the {tag.Model} model, which the poll planner does not support. " +
-               "v1 plans KeyValue and WideRow only; the named-query model is deferred (design §5.4).");
+        => new($"Sql tag '{tag.Name}' uses the {tag.Model} model, which the poll planner does not support.");
 }

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs

@@ -52,6 +52,9 @@ public sealed class SqlPollReader : IDisposable
     private readonly Func<string, SqlTagDefinition?> _resolve;
     private readonly ILogger _logger;
 
+    /// <summary>The driver's validated named queries, for <see cref="SqlTagModel.Query"/> plans.</summary>
+    private readonly IReadOnlyDictionary<string, SqlNamedQuery>? _queries;
+
     /// <summary>
     ///     Caps concurrently executing group queries — and therefore concurrently held connections
     ///     (design §8.4). Disposed by <see cref="Dispose"/> only once every group that ever acquired it has
@@ -132,6 +135,11 @@ public sealed class SqlPollReader : IDisposable
     ///     straight through.
     /// </param>
     /// <param name="logger">Optional; defaults to a no-op logger.</param>
+    /// <param name="queries">
+    ///     The driver's validated named queries (design §5.4); null when it declares none, in which case a
+    ///     <see cref="SqlTagModel.Query"/> definition cannot be planned and publishes
+    ///     <see cref="SqlStatusCodes.BadConfigurationError"/>.
+    /// </param>
     /// <exception cref="ArgumentNullException">A required reference argument is null.</exception>
     /// <exception cref="ArgumentException"><paramref name="connectionString"/> is blank.</exception>
     /// <exception cref="ArgumentOutOfRangeException">A timeout is non-positive, or the cap is below 1.</exception>
@@ -144,7 +152,8 @@ public sealed class SqlPollReader : IDisposable
         int maxConcurrentGroups,
         bool nullIsBad,
         Func<string, SqlTagDefinition?> resolve,
-        ILogger? logger = null)
+        ILogger? logger = null,
+        IReadOnlyDictionary<string, SqlNamedQuery>? queries = null)
     {
         ArgumentNullException.ThrowIfNull(factory);
         ArgumentNullException.ThrowIfNull(dialect);
@@ -163,6 +172,7 @@ public sealed class SqlPollReader : IDisposable
         _nullIsBad = nullIsBad;
         _resolve = resolve;
         _logger = logger ?? NullLogger.Instance;
+        _queries = queries;
         _gate = new SemaphoreSlim(maxConcurrentGroups, maxConcurrentGroups);
     }
 
@@ -276,7 +286,7 @@ public sealed class SqlPollReader : IDisposable
 
         try
         {
-            return new PollShape(resolved, definitions, slots, SqlGroupPlanner.Plan(definitions, _dialect), PlanningFailed: false);
+            return new PollShape(resolved, definitions, slots, SqlGroupPlanner.Plan(definitions, _dialect, _queries), PlanningFailed: false);
         }
         catch (Exception ex) when (ex is ArgumentException or NotSupportedException)
         {
@@ -516,7 +526,7 @@ public sealed class SqlPollReader : IDisposable
             _logger.LogWarning(
                 "Sql poll group ({Model}, '{Table}', {Members} tag(s)) exceeded the {Timeout} ms operation " +
                 "timeout; publishing BadTimeout. Repeats for this group are suppressed for {Interval} s.",
-                plan.Model, plan.Members[0].Table, plan.Members.Count,
+                plan.Model, DescribeSource(plan), plan.Members.Count,
                 (int)_operationTimeout.TotalMilliseconds, (int)ContractWarningInterval.TotalSeconds);
         }
 
@@ -546,12 +556,25 @@ public sealed class SqlPollReader : IDisposable
 
             try
             {
+                // A named query is operator-authored SQL. The authoring guard (SqlReadOnlyStatement) already
+                // refused anything that is not a single SELECT, and this is the backstop behind it: the
+                // statement runs inside a transaction that is ALWAYS rolled back, so a side effect the guard
+                // could not see (a function that writes) is never committed. Disposing the transaction without
+                // a commit rolls it back, including on the timeout/abandon path.
+                var transaction = plan.Model == SqlTagModel.Query
+                    ? await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false)
+                    : null;
+                await using var rollback = transaction is null
+                    ? null
+                    : new RollbackOnDispose(transaction);
+
                 var command = connection.CreateCommand();
                 await using (command.ConfigureAwait(false))
                 {
                     // Before any parameter is added: on ODP.NET this is what turns bind-by-position off, and
                     // a positionally-bound IN list feeds one tag's value to another tag's node in silence.
                     _dialect.ConfigureCommand(command);
+                    command.Transaction = transaction;
                     command.CommandText = plan.SqlText;
                     command.CommandTimeout = _commandTimeoutSeconds;
                     for (var i = 0; i < plan.ParameterBindNames.Count; i++)
@@ -583,9 +606,10 @@ public sealed class SqlPollReader : IDisposable
                 if (ShouldWarn("failed:" + plan.SqlText))
                 {
                     _logger.LogWarning(
-                        ex, "Sql poll group ({Model}, {Members} tag(s)) failed: {Sql}. Repeats for this group are " +
-                        "suppressed for {Interval} s.",
-                        plan.Model, plan.Members.Count, plan.SqlText, (int)ContractWarningInterval.TotalSeconds);
+                        ex, "Sql poll group ({Model}, {Source}, {Members} tag(s)) failed: {Sql}. Repeats for this " +
+                        "group are suppressed for {Interval} s.",
+                        plan.Model, DescribeSource(plan), plan.Members.Count, plan.SqlText,
+                        (int)ContractWarningInterval.TotalSeconds);
                 }
                 return GroupResult.Failed;
             }
@@ -600,7 +624,11 @@ public sealed class SqlPollReader : IDisposable
     private async Task<GroupResult> MaterialiseAsync(
         SqlQueryPlan plan, DbDataReader reader, CancellationToken cancellationToken)
     {
-        var selected = plan.SelectedColumns;
+        // A named query's projection is the operator's, not the planner's: read every result column, by the
+        // name the result set reports. A duplicated name keeps its first ordinal.
+        IReadOnlyList<string> selected = plan.Model == SqlTagModel.Query
+            ? ResultColumnNames(reader)
+            : plan.SelectedColumns;
         var ordinals = new int[selected.Count];
         var typeNames = new string[selected.Count];
         var columnIndex = new Dictionary<string, int>(selected.Count, StringComparer.Ordinal);
@@ -609,9 +637,9 @@ public sealed class SqlPollReader : IDisposable
             // GetOrdinal rather than the position in SelectedColumns: the list is documented to be in
             // ordinal order, but resolving it against the live result set is what makes a wrong slice
             // impossible rather than merely unlikely.
-            ordinals[c] = reader.GetOrdinal(selected[c]);
+            ordinals[c] = plan.Model == SqlTagModel.Query ? c : reader.GetOrdinal(selected[c]);
             typeNames[c] = SafeTypeName(reader, ordinals[c]);
-            columnIndex[selected[c]] = c;
+            columnIndex.TryAdd(selected[c], c);
         }
 
         // The client-side half of the row cap (R06:S-1): the statement already asks for RowCap + 1 rows, but a
@@ -633,15 +661,20 @@ public sealed class SqlPollReader : IDisposable
             if (ShouldWarn("rowcap:" + plan.SqlText))
             {
                 var first = plan.Members[0];
-                var selector = plan.Model == SqlTagModel.KeyValue
-                    ? string.Concat(first.KeyColumn, " IN (", (rowCap / 2).ToString(CultureInfo.InvariantCulture), " key(s))")
-                    : string.Concat(first.RowSelectorColumn, " = ", first.RowSelectorValue);
+                var selector = plan.Model switch
+                {
+                    SqlTagModel.KeyValue => string.Concat(
+                        first.KeyColumn, " IN (", (rowCap / 2).ToString(CultureInfo.InvariantCulture), " key(s))"),
+                    SqlTagModel.Query => "maxRows",
+                    _ => string.Concat(first.RowSelectorColumn, " = ", first.RowSelectorValue),
+                };
                 _logger.LogWarning(
                     "Sql poll source '{Table}' returned more rows than the {Model} selector '{Selector}' can attribute " +
                     "(at most {Cap}) in one poll, so the selector is ambiguous — a history/events table, or a " +
                     "where column that is not unique. Its {Members} tag(s) publish BadConfigurationError and " +
-                    "nothing past row {Limit} is read. Point the tag at a current-values table or a latest-row view.",
-                    first.Table, plan.Model, selector, rowCap, plan.Members.Count, rowCap + 1);
+                    "nothing past row {Limit} is read. Point the tag at a current-values table or a latest-row view " +
+                    "(for a named query: narrow its WHERE, or raise its maxRows).",
+                    DescribeSource(plan), plan.Model, selector, rowCap, plan.Members.Count, rowCap + 1);
             }
 
             return GroupResult.RowCapExceeded;
@@ -699,6 +732,11 @@ public sealed class SqlPollReader : IDisposable
             {
                 snapshot = new DataValueSnapshot(null, outcome.FailureStatus, null, readAt);
             }
+            else if (plan.Model == SqlTagModel.Query)
+            {
+                snapshot = MapQueryMember(plan, member, outcome, readAt, out var refused);
+                refusedSome |= refused;
+            }
             else if (duplicatedKeys is not null && member.KeyValue is { } key && duplicatedKeys.Contains(key))
             {
                 // More than one row for THIS tag's key: its selector is ambiguous — the same rule a where-pair
@@ -834,6 +872,105 @@ public sealed class SqlPollReader : IDisposable
         // No row: the key was deleted, or the row selector matched nothing. Distinct from a NULL cell.
         if (row is null) return new DataValueSnapshot(null, SqlStatusCodes.BadNoData, null, readAt);
 
+        return MapCell(member, outcome, row, valueColumn, timestampColumn, readAt);
+    }
+
+    /// <summary>
+    ///     Maps one <see cref="SqlTagModel.Query"/> member: resolves its <c>column</c> (and <c>rowKey</c> /
+    ///     <c>timestampColumn</c>) against the result set's own column names, picks its row, and maps the cell.
+    ///     <list type="bullet">
+    ///         <item>A <c>column</c> or <c>rowKey.column</c> the result does not carry is an authoring fault the
+    ///         database just proved: <see cref="SqlStatusCodes.BadConfigurationError"/>, counted as answered.</item>
+    ///         <item>No row for the member (an empty result, or no row whose key matches): <see cref="SqlStatusCodes.BadNoData"/>.</item>
+    ///         <item>More than one candidate row (no rowKey and a multi-row result, or a duplicated key): the
+    ///         selector is ambiguous, so <see cref="SqlStatusCodes.BadConfigurationError"/> — never "last row wins".</item>
+    ///     </list>
+    ///     Column names match exactly first, then by a unique case-insensitive match (Postgres folds an unquoted
+    ///     alias to lower case, Oracle to upper case), the same rule the catalog gate applies to identifiers.
+    /// </summary>
+    private DataValueSnapshot MapQueryMember(
+        SqlQueryPlan plan, SqlTagDefinition member, GroupResult outcome, DateTime readAt, out bool refused)
+    {
+        refused = false;
+        var columns = outcome.ColumnIndex.Keys.ToArray();
+        if (!SqlCatalog.TryMatch(columns, member.ColumnName ?? string.Empty, out var valueColumn))
+        {
+            refused = true;
+            WarnQueryColumnMissing(plan, member, member.ColumnName, columns);
+            return new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
+        }
+
+        object?[]? row;
+        if (member.RowKeyColumn is { } keyColumn)
+        {
+            if (!SqlCatalog.TryMatch(columns, keyColumn, out var resolvedKey))
+            {
+                refused = true;
+                WarnQueryColumnMissing(plan, member, keyColumn, columns);
+                return new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
+            }
+
+            var keyAt = outcome.ColumnIndex[resolvedKey];
+            row = null;
+            var matches = 0;
+            foreach (var candidate in outcome.Rows)
+            {
+                // Matched as text, exactly as a key-value key is: the authored value is configuration text.
+                if (candidate[keyAt] is not { } cell
+                    || !string.Equals(Convert.ToString(cell, CultureInfo.InvariantCulture), member.RowKeyValue,
+                        StringComparison.Ordinal))
+                    continue;
+                row = candidate;
+                matches++;
+            }
+
+            if (matches > 1)
+            {
+                refused = true;
+                return new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
+            }
+        }
+        else
+        {
+            if (outcome.Rows.Count > 1)
+            {
+                refused = true;
+                return new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
+            }
+
+            row = outcome.Rows.Count == 1 ? outcome.Rows[0] : null;
+        }
+
+        if (row is null) return new DataValueSnapshot(null, SqlStatusCodes.BadNoData, null, readAt);
+
+        string? timestampColumn = null;
+        if (!string.IsNullOrWhiteSpace(member.TimestampColumn)
+            && SqlCatalog.TryMatch(columns, member.TimestampColumn, out var resolvedTimestamp))
+            timestampColumn = resolvedTimestamp;
+
+        return MapCell(member, outcome, row, valueColumn, timestampColumn, readAt);
+    }
+
+    /// <summary>Rate-limited warning for a Query tag naming a column its query does not project.</summary>
+    private void WarnQueryColumnMissing(
+        SqlQueryPlan plan, SqlTagDefinition member, string? column, IReadOnlyList<string> returned)
+    {
+        if (!ShouldWarn("querycolumn:" + plan.SqlText + "\0" + column)) return;
+        _logger.LogWarning(
+            "Sql tag '{Tag}' reads column '{Column}', which named query '{Query}' does not return (or matches more " +
+            "than one returned column by case); it publishes BadConfigurationError. Returned columns: {Columns}.",
+            member.Name, column, member.QueryName, string.Join(", ", returned));
+    }
+
+    /// <summary>Maps one present row's cell to a snapshot: value, quality, and source timestamp.</summary>
+    private DataValueSnapshot MapCell(
+        SqlTagDefinition member,
+        GroupResult outcome,
+        object?[] row,
+        string? valueColumn,
+        string? timestampColumn,
+        DateTime readAt)
+    {
         // A timestamp that will not parse falls back to the poll clock rather than poisoning a good value.
         var sourceTimestamp = ReadTimestamp(outcome, row, timestampColumn) ?? readAt;
 
@@ -931,6 +1068,44 @@ public sealed class SqlPollReader : IDisposable
             Convert.ToDateTime(cell, CultureInfo.InvariantCulture), DateTimeKind.Utc),
     };
 
+    /// <summary>The operator-facing name of what a plan reads — its table, or its named query.</summary>
+    private static string DescribeSource(SqlQueryPlan plan)
+        => plan.Model == SqlTagModel.Query
+            ? string.Concat("query ", plan.Members[0].QueryName)
+            : plan.Members[0].Table;
+
+    /// <summary>Every result column's name, in ordinal order.</summary>
+    private static string[] ResultColumnNames(DbDataReader reader)
+    {
+        var names = new string[reader.FieldCount];
+        for (var i = 0; i < names.Length; i++) names[i] = reader.GetName(i);
+        return names;
+    }
+
+    /// <summary>
+    ///     Rolls a named query's transaction back when disposed — always, because a named query is a read and
+    ///     nothing it did may be committed. A rollback that fails (the connection already broke) is swallowed:
+    ///     the uncommitted work dies with the connection either way.
+    /// </summary>
+    private sealed class RollbackOnDispose(DbTransaction transaction) : IAsyncDisposable
+    {
+        public async ValueTask DisposeAsync()
+        {
+            try
+            {
+                await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false);
+            }
+            catch (Exception ex) when (ex is DbException or InvalidOperationException)
+            {
+                // Nothing was committed; the transaction ends with its connection.
+            }
+            finally
+            {
+                await transaction.DisposeAsync().ConfigureAwait(false);
+            }
+        }
+    }
+
     /// <summary>Rate-limits one per-poll warning by key; true at most once per <see cref="ContractWarningInterval"/>.</summary>
     private bool ShouldWarn(string key)
     {

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlQueryPlan.cs

@@ -53,13 +53,15 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
 /// </param>
 /// <param name="SelectedColumns">
 ///     The bare (unquoted) column names in <see cref="SqlText"/>'s SELECT list, in order and distinct — the
-///     reader's map from result-set ordinal to source column.
+///     reader's map from result-set ordinal to source column. Empty for a <see cref="SqlTagModel.Query"/> plan,
+///     whose projection is the named query's own: the reader reads every result column by its reported name.
 /// </param>
 /// <param name="RowCap">
 ///     The most rows the reader will attribute before refusing the whole group — twice the distinct key count
 ///     (key-value: headroom so stray duplicates are attributed per key) or 1 (a where-pair wide row: one
 ///     selector, so any second row is the group's ambiguity); <see langword="null"/> when the statement
-///     already limits itself (topByTimestamp). <see cref="SqlText"/> asks for <c>RowCap + 1</c>, and the
+///     already limits itself (topByTimestamp). For a <see cref="SqlTagModel.Query"/> plan it is the query's
+///     <c>maxRows</c>, enforced client-side only. <see cref="SqlText"/> asks for <c>RowCap + 1</c>, and the
 ///     reader stops there: receiving the extra row means a history/events table or a non-unique where column,
 ///     and the group publishes <c>BadConfigurationError</c> instead of materialising every matching row on
 ///     every poll (R06:S-1).

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/SqlDriverForm.razor

@@ -97,6 +97,25 @@
     </div>
 </section>
 
+@* Named queries (design §5.4) — read by Query-model tags *@
+<section class="panel rise mt-3" style="animation-delay:.12s">
+    <div class="panel-head">Named queries</div>
+    <div style="padding:1rem">
+        <textarea class="form-control form-control-sm mono" rows="6" value="@_form.QueriesJson"
+                  placeholder='{ "activeBatch": { "sql": "SELECT line, speed, ts FROM dbo.LineStatus WHERE active = @@active", "params": { "active": true } } }'
+                  @onchange="@(async e => { _form.SetQueriesJson(e.Value?.ToString()); await EmitAsync(); })"></textarea>
+        <div class="form-text">
+            JSON object: name → <code>{ "sql", "params", "maxRows" }</code>. Each statement must be a single read-only
+            <code>SELECT</code>/<code>WITH</code>; parameters are always bound (write the provider's marker — <code>@@name</code>,
+            Oracle <code>:p_name</code>, Odbc <code>?</code> in order). Queries run inside a rolled-back transaction.
+        </div>
+        @if (_form.QueriesError is { } queriesError)
+        {
+            <div class="text-danger small mt-1">@queriesError — the last valid queries are kept until this is fixed.</div>
+        }
+    </div>
+</section>
+
 <DriverResilienceSection ResilienceConfig="@ResilienceConfig" ResilienceConfigChanged="OnResilienceChanged" />
 
 @code {
@@ -171,6 +190,57 @@
         public int? MaxConcurrentGroups { get; set; }
         public bool NullIsBad { get; set; }
 
+        /// <summary>The named queries the form emits — the last text that parsed and validated.</summary>
+        public Dictionary<string, SqlNamedQueryDto>? Queries { get; set; }
+
+        /// <summary>The operator's queries text, as typed.</summary>
+        public string QueriesJson { get; set; } = "";
+
+        /// <summary>Why <see cref="QueriesJson"/> was refused, or null when it is in force.</summary>
+        public string? QueriesError { get; private set; }
+
+        private static readonly JsonSerializerOptions QueryJsonOptions = new()
+        {
+            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
+            PropertyNameCaseInsensitive = true,
+            DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
+            WriteIndented = true,
+        };
+
+        /// <summary>Adopts edited queries text when it parses and every entry passes the runtime's own rules
+        /// (<see cref="SqlNamedQueryRules.Describe"/>); otherwise records why and keeps the last valid set, so a
+        /// typo never silently drops queries the deployed tags read.</summary>
+        public void SetQueriesJson(string? text)
+        {
+            QueriesJson = text ?? "";
+            QueriesError = TryParseQueries(QueriesJson, out var parsed);
+            if (QueriesError is null) { Queries = parsed; }
+        }
+
+        /// <summary>Parses + validates queries text. Blank ⇒ no queries. Returns the problem, or null.</summary>
+        public static string? TryParseQueries(string? text, out Dictionary<string, SqlNamedQueryDto>? queries)
+        {
+            queries = null;
+            if (string.IsNullOrWhiteSpace(text)) { return null; }
+            try
+            {
+                queries = JsonSerializer.Deserialize<Dictionary<string, SqlNamedQueryDto>>(text, QueryJsonOptions);
+            }
+            catch (JsonException ex)
+            {
+                return $"named queries are not valid JSON (line {(ex.LineNumber ?? 0) + 1})";
+            }
+
+            if (queries is null) { return "named queries must be a JSON object"; }
+            foreach (var (name, query) in queries)
+            {
+                if (SqlNamedQueryRules.Describe(name, query) is { } problem) { queries = null; return problem; }
+            }
+
+            if (queries.Count == 0) { queries = null; }
+            return null;
+        }
+
         public static FormModel FromDto(SqlDriverConfigDto d) => new()
         {
             Provider = d.Provider,
@@ -180,6 +250,8 @@
             CommandTimeoutSeconds = ToSeconds(d.CommandTimeout),
             MaxConcurrentGroups = d.MaxConcurrentGroups,
             NullIsBad = d.NullIsBad ?? false,
+            Queries = d.Queries,
+            QueriesJson = d.Queries is { Count: > 0 } q ? JsonSerializer.Serialize(q, QueryJsonOptions) : "",
         };
 
         public SqlDriverConfigDto ToDto() => new()
@@ -193,6 +265,7 @@
             // Emit nullIsBad only when true; unchecked ⇒ null ⇒ the factory default (Uncertain). allowWrites
             // and rawTags are deliberately left null so they are omitted from the authored blob.
             NullIsBad = NullIsBad ? true : null,
+            Queries = Queries is { Count: > 0 } ? Queries : null,
         };
 
         private static int? ToSeconds(TimeSpan? ts) => ts.HasValue ? (int)ts.Value.TotalSeconds : null;

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/SqlTagConfigEditor.razor

@@ -1,4 +1,4 @@
-@* Typed TagConfig editor for the read-only Sql driver. Thin shell over SqlTagConfigModel: every field edit
+@* Typed TagConfig editor for the Sql driver. Thin shell over SqlTagConfigModel: every field edit
    goes model → ToJson → ConfigJsonChanged (the model is the single source of the blob shape + enum-name
    serialisation; never hand-composed here). The Model dropdown (KeyValue/WideRow) shows the matching field
    group; the "Build address" picker is the ASSISTED path — every field is also hand-editable. Dispatched
@@ -14,12 +14,21 @@
 <div class="row g-2">
     <div class="col-md-4"><label class="form-label">Model</label>
         <select class="form-select form-select-sm" value="@_m.Model" @onchange="@(e => Update(() => _m.Model = ParseEnum(e.Value, SqlTagModel.KeyValue)))">
-            @* Query is deferred (P3) and rejected by the parser/validator — only offer the two v1 models. *@
             <option value="@SqlTagModel.KeyValue">KeyValue</option>
             <option value="@SqlTagModel.WideRow">WideRow</option>
+            <option value="@SqlTagModel.Query">Query (named query)</option>
         </select></div>
-    <div class="col-md-5"><label class="form-label">Table or view</label>
-        <input class="form-control form-control-sm mono" value="@_m.Table" placeholder="dbo.TagValues" @onchange="@(e => Update(() => _m.Table = NullIfBlank(e.Value)))" /></div>
+    @if (_m.Model != SqlTagModel.Query)
+    {
+        <div class="col-md-5"><label class="form-label">Table or view</label>
+            <input class="form-control form-control-sm mono" value="@_m.Table" placeholder="dbo.TagValues" @onchange="@(e => Update(() => _m.Table = NullIfBlank(e.Value)))" /></div>
+    }
+    else
+    {
+        <div class="col-md-5"><label class="form-label">Named query</label>
+            <input class="form-control form-control-sm mono" value="@_m.QueryName" placeholder="activeBatch" @onchange="@(e => Update(() => _m.QueryName = NullIfBlank(e.Value)))" />
+            <div class="form-text">A key of the driver config's <code>queries</code>.</div></div>
+    }
     <div class="col-md-3"><label class="form-label">Type override</label>
         <select class="form-select form-select-sm" value="@(_m.Type?.ToString() ?? "")" @onchange="@(e => Update(() => _m.Type = ParseNullableEnum<DriverDataType>(e.Value)))">
             <option value="">(infer from column)</option>
@@ -27,7 +36,20 @@
         </select></div>
 </div>
 
-@if (_m.Model == SqlTagModel.KeyValue)
+@if (_m.Model == SqlTagModel.Query)
+{
+    <div class="row g-2 mt-1">
+        <div class="col-md-4"><label class="form-label">Result column</label>
+            <input class="form-control form-control-sm mono" value="@_m.Column" placeholder="speed" @onchange="@(e => Update(() => _m.Column = NullIfBlank(e.Value)))" />
+            <div class="form-text">The column of the query's result this tag reads.</div></div>
+        <div class="col-md-4"><label class="form-label">Row key column <span class="text-muted small">(optional)</span></label>
+            <input class="form-control form-control-sm mono" value="@_m.RowKey.Column" placeholder="line" @onchange="@(e => Update(() => SetRowKey(NullIfBlank(e.Value), _m.RowKey.Value)))" /></div>
+        <div class="col-md-4"><label class="form-label">Row key value</label>
+            <input class="form-control form-control-sm mono" value="@_m.RowKey.Value" placeholder="Line1" @onchange="@(e => Update(() => SetRowKey(_m.RowKey.Column, e.Value?.ToString())))" />
+            <div class="form-text">Both blank ⇒ the query must return exactly one row.</div></div>
+    </div>
+}
+else if (_m.Model == SqlTagModel.KeyValue)
 {
     <div class="row g-2 mt-1">
         <div class="col-md-4"><label class="form-label">Key column</label>
@@ -158,6 +180,14 @@ else
         return string.IsNullOrWhiteSpace(s) ? null : s;
     }
 
+    // Both row-key fields blank clears the rowKey entirely, so the blob carries no half-authored selector.
+    private void SetRowKey(string? column, string? value)
+    {
+        if (string.IsNullOrWhiteSpace(column) && string.IsNullOrEmpty(value)) { _m.RowKey.Clear(); return; }
+        _m.RowKey.Column = column;
+        _m.RowKey.Value = value;
+    }
+
     private async Task Update(Action apply)
     {
         apply();

src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Uns/TagEditors/SqlTagConfigModel.cs

@@ -6,7 +6,7 @@ using RuntimeTagConfigJson = ZB.MOM.WW.OtOpcUa.Core.Abstractions.TagConfigJson;
 namespace ZB.MOM.WW.OtOpcUa.AdminUI.Uns.TagEditors;
 
 /// <summary>
-/// Typed working model for a read-only Sql tag's TagConfig JSON (the driver-specific table/column
+/// Typed working model for a Sql tag's TagConfig JSON (the driver-specific table/column
 /// mapping; name/writable live on the Tag entity). Mirrors — byte-for-byte on the fields it owns — the
 /// authoritative runtime reader <see cref="SqlEquipmentTagParser.TryParse"/>: the <c>model</c> and
 /// <c>type</c> enums are written as their NAME strings (never numbers), and <see cref="Validate"/>
@@ -15,11 +15,12 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Uns.TagEditors;
 /// </summary>
 public sealed class SqlTagConfigModel
 {
-    /// <summary>Tag→value mapping model. v1 accepts <see cref="SqlTagModel.KeyValue"/> and
-    /// <see cref="SqlTagModel.WideRow"/>; <see cref="SqlTagModel.Query"/> is deferred and rejected.</summary>
+    /// <summary>Tag→value mapping model: <see cref="SqlTagModel.KeyValue"/>, <see cref="SqlTagModel.WideRow"/>
+    /// or <see cref="SqlTagModel.Query"/> (a driver-config named query).</summary>
     public SqlTagModel Model { get; set; } = SqlTagModel.KeyValue;
 
-    /// <summary>The table or view to read (e.g. <c>dbo.TagValues</c>). Required for every model.</summary>
+    /// <summary>The table or view to read (e.g. <c>dbo.TagValues</c>). Required for KeyValue and WideRow; a
+    /// Query tag reads its named query instead.</summary>
     public string? Table { get; set; }
 
     /// <summary><see cref="SqlTagModel.KeyValue"/>: the column matched against <see cref="KeyValue"/>.</summary>
@@ -42,6 +43,17 @@ public sealed class SqlTagConfigModel
     /// <summary><see cref="SqlTagModel.WideRow"/>: which row to read — a where-pair or newest-by-timestamp.</summary>
     public SqlRowSelectorModel RowSelector { get; set; } = new();
 
+    /// <summary><see cref="SqlTagModel.Query"/>: the driver-config <c>queries</c> entry this tag reads (the
+    /// blob's <c>query</c> key).</summary>
+    public string? QueryName { get; set; }
+
+    /// <summary><see cref="SqlTagModel.Query"/>: the result-set column this tag reads (the blob's <c>column</c>).</summary>
+    public string? Column { get; set; }
+
+    /// <summary><see cref="SqlTagModel.Query"/>: which result row this tag reads. Empty ⇒ the query must return
+    /// exactly one row.</summary>
+    public SqlRowKeyModel RowKey { get; set; } = new();
+
     /// <summary>Optional explicit OPC UA data type override (the blob's <c>type</c> field). <c>null</c> ⇒ the
     /// driver infers the type from the result-set column metadata.</summary>
     public DriverDataType? Type { get; set; }
@@ -79,6 +91,9 @@ public sealed class SqlTagConfigModel
             TimestampColumn = TagConfigJson.GetString(o, "timestampColumn"),
             ColumnName = TagConfigJson.GetString(o, "columnName"),
             RowSelector = SqlRowSelectorModel.FromJson(selectorObj),
+            QueryName = TagConfigJson.GetString(o, "query"),
+            Column = TagConfigJson.GetString(o, "column"),
+            RowKey = SqlRowKeyModel.FromJson(o.TryGetPropertyValue("rowKey", out var rk) ? rk : null),
             Type = type,
             _bag = o,
             _source = json ?? "",
@@ -101,6 +116,12 @@ public sealed class SqlTagConfigModel
         TagConfigJson.Set(_bag, "timestampColumn", TimestampColumn);
         TagConfigJson.Set(_bag, "columnName", ColumnName);
         TagConfigJson.Set(_bag, "type", Type);
+        TagConfigJson.Set(_bag, "query", QueryName);
+        TagConfigJson.Set(_bag, "column", Column);
+
+        var rowKey = RowKey.ToJsonNode();
+        if (rowKey is null) { _bag.Remove("rowKey"); }
+        else { _bag["rowKey"] = rowKey; }
 
         var selector = RowSelector.ToJsonObject();
         if (selector is null) { _bag.Remove("rowSelector"); }
@@ -118,8 +139,23 @@ public sealed class SqlTagConfigModel
         if (_invalidModelRaw is not null) { return $"model: '{_invalidModelRaw}' is not a valid Sql tag model."; }
         if (_invalidTypeRaw is not null) { return $"type: '{_invalidTypeRaw}' is not a valid data type."; }
 
-        // Query is design §5.4 / P3 — the parser rejects it; the editor must too.
-        if (Model == SqlTagModel.Query) { return "model 'Query' is deferred (P3) and not supported."; }
+        if (Model == SqlTagModel.Query)
+        {
+            // Mirrors the parser's Query arm: query + column are required, and a rowKey — when present at all —
+            // needs both its column and a scalar value. Table is not read (the named query is the source).
+            if (string.IsNullOrWhiteSpace(QueryName)) { return "query is required for a Query tag (the driver-config named query)."; }
+            if (string.IsNullOrWhiteSpace(Column)) { return "column is required for a Query tag (the result-set column)."; }
+            if (RowKey.Present && (string.IsNullOrWhiteSpace(RowKey.Column) || RowKey.Value is null))
+            {
+                return "rowKey needs both a column and a value (clear it when the query returns exactly one row).";
+            }
+
+            return TagConfigParseGate.Describe(
+                _source,
+                j => SqlEquipmentTagParser.TryParse(j, TagConfigParseGate.AuthoringRawPath, out _),
+                SqlTagDefinitionFactory.Inspect,
+                "Sql");
+        }
 
         if (string.IsNullOrWhiteSpace(Table)) { return "table is required."; }
 
@@ -224,3 +260,60 @@ public sealed class SqlRowSelectorModel
         return o.Count == 0 ? null : o;
     }
 }
+
+/// <summary>
+/// Typed working model for a <see cref="SqlTagModel.Query"/> tag's optional nested <c>rowKey</c> object
+/// (<c>{"column": …, "value": …}</c>). <see cref="Present"/> records whether the blob carried the key at all, so a
+/// half-authored rowKey is reported rather than silently read as "no rowKey". Preserves unknown keys.
+/// </summary>
+public sealed class SqlRowKeyModel
+{
+    /// <summary>The result-set column that picks the row.</summary>
+    public string? Column { get; set; }
+
+    /// <summary>The value <see cref="Column"/> must hold (matched as text). Captured as text from a JSON string,
+    /// number or boolean, mirroring the parser's scalar read.</summary>
+    public string? Value { get; set; }
+
+    /// <summary>True when the blob carried a <c>rowKey</c> (object or not) or a field here has been set.</summary>
+    public bool Present => _authored || Column is not null || Value is not null;
+
+    private bool _authored;
+    private JsonObject _bag = new();
+
+    /// <summary>Loads from the blob's <c>rowKey</c> node, or an empty model when absent.</summary>
+    /// <param name="node">The <c>rowKey</c> node, or <c>null</c> when absent.</param>
+    /// <returns>The populated model.</returns>
+    public static SqlRowKeyModel FromJson(JsonNode? node)
+    {
+        if (node is null) { return new SqlRowKeyModel(); }
+        var bag = node is JsonObject o ? o.DeepClone().AsObject() : new JsonObject();
+        return new SqlRowKeyModel
+        {
+            Column = TagConfigJson.GetString(bag, "column"),
+            Value = TagConfigJson.GetScalarText(bag, "value"),
+            _bag = bag,
+            // A non-object rowKey is still "present" — the parser refuses it, and so must Validate.
+            _authored = true,
+        };
+    }
+
+    /// <summary>Clears the row key, so <see cref="ToJsonNode"/> omits it.</summary>
+    public void Clear()
+    {
+        Column = null;
+        Value = null;
+        _authored = false;
+        _bag = new JsonObject();
+    }
+
+    /// <summary>Rebuilds the <c>rowKey</c> object over the preserved bag, or <c>null</c> when empty.</summary>
+    /// <returns>The node, or null.</returns>
+    public JsonObject? ToJsonNode()
+    {
+        var o = _bag.DeepClone().AsObject();
+        TagConfigJson.Set(o, "column", Column);
+        TagConfigJson.Set(o, "value", Value);
+        return o.Count == 0 && !_authored ? null : o;
+    }
+}

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs

@@ -466,11 +466,11 @@ public sealed class SqlDriverTests
 
     /// <summary>
     ///     <b>R06:S-4.</b> A skipped tag's ledger reason (what /hosts shows) is the deploy gate's own explanation
-    ///     — for a <c>Query</c> tag, that the model is deferred — instead of a generic "did not map". Control: a
-    ///     well-formed tag is not skipped at all.
+    ///     — for an incomplete <c>Query</c> tag, the missing field — instead of a generic "did not map". Control:
+    ///     a well-formed tag is not skipped at all.
     /// </summary>
     [Fact]
-    public async Task ADeferredQueryTag_isSkipped_withTheGatesReason()
+    public async Task AnIncompleteQueryTag_isSkipped_withTheGatesReason()
     {
         using var fixture = new SqlitePollFixture();
         await using var driver = NewDriver(fixture,
@@ -482,6 +482,7 @@ public sealed class SqlDriverTests
         var skipped = report.Sample.ShouldHaveSingleItem();
         skipped.RawPath.ShouldBe("Q");
         skipped.Reason.ShouldContain("Query");
+        skipped.Reason.ShouldContain("'column'");
     }
 
     /// <summary>

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlEquipmentTagParserTests.cs

@@ -225,23 +225,24 @@ public class SqlEquipmentTagParserTests
             "raw", out _).ShouldBeFalse();
 
     [Fact]
-    public void TryParse_theDeferredQueryModel_rejectsTag()
+    public void TryParse_aQueryTagMissingItsColumn_rejectsTag()
     {
-        // Design §5.4 / P3. Serving it would materialise a node the runtime has no way to read.
+        // A Query tag names the driver-config query AND the result column; the old P3-era blob (a table + an
+        // inline SQL "query") carries no column, so it is still refused.
         const string json = """{"driver":"Sql","model":"Query","table":"T","query":"SELECT 1"}""";
         SqlEquipmentTagParser.TryParse(json, "raw", out _).ShouldBeFalse();
     }
 
     /// <summary>
     ///     <b>R06:S-4.</b> The deploy gate (<c>EquipmentTagConfigInspector</c>) reaches this through the parser:
-    ///     a <c>Query</c> tag draws a warning naming the deferral (a rejection under TagConfigValidation=Error).
-    ///     Control: a served key-value blob draws nothing.
+    ///     an incomplete <c>Query</c> tag draws a warning naming the missing field (a rejection under
+    ///     TagConfigValidation=Error). Control: a served key-value blob draws nothing.
     /// </summary>
     [Fact]
-    public void Inspect_theDeferredQueryModel_isNamedAtTheDeployGate()
+    public void Inspect_anIncompleteQueryTag_isNamedAtTheDeployGate()
     {
         SqlTagDefinitionFactory.Inspect("""{"driver":"Sql","model":"Query","table":"T","query":"SELECT 1"}""")
-            .ShouldHaveSingleItem().ShouldContain("Query");
+            .ShouldHaveSingleItem().ShouldContain("'column'");
         SqlTagDefinitionFactory.Inspect(
                 """{"driver":"Sql","model":"KeyValue","table":"T","keyColumn":"k","keyValue":"v","valueColumn":"c"}""")
             .ShouldBeEmpty();

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlGroupPlannerTests.cs

@@ -482,10 +482,10 @@ public class SqlGroupPlannerTests
     }
 
     [Fact]
-    public void TheDeferredQueryModel_throws_ratherThanBeingSilentlyDropped()
+    public void AQueryTagWithNoNamedQueries_throws_ratherThanBeingSilentlyDropped()
     {
-        var tag = new SqlTagDefinition("A", SqlTagModel.Query, "dbo.T");
-        Should.Throw<NotSupportedException>(() => SqlGroupPlanner.Plan([tag], new SqlServerDialect()));
+        var tag = new SqlTagDefinition("A", SqlTagModel.Query, string.Empty, ColumnName: "c", QueryName: "q");
+        Should.Throw<ArgumentException>(() => SqlGroupPlanner.Plan([tag], new SqlServerDialect()));
     }
 
     [Fact]

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlNamedQueryTests.cs

@@ -0,0 +1,370 @@
+using System.Data;
+using System.Data.Common;
+using System.Globalization;
+using System.Text.Json.Nodes;
+using Microsoft.Data.Sqlite;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;
+
+/// <summary>
+///     The named-query model (<see cref="SqlTagModel.Query"/>, design §5.4, lmxopcua#753): the tag blob parse,
+///     the read-only statement guard, the factory's query validation, the planner, and the end-to-end read
+///     through <see cref="SqlDriver"/> over SQLite — including the rolled-back transaction every named query
+///     runs inside.
+/// </summary>
+public sealed class SqlNamedQueryTests
+{
+    private const string DriverInstanceId = "sql-query";
+
+    // ---- tag blob parse + deploy-gate explanation ----
+
+    [Fact]
+    public void TryParse_aQueryTag_mapsQueryColumnRowKeyAndType()
+    {
+        const string json = """
+            {"driver":"Sql","model":"Query","query":"lineStatus","column":"speed",
+             "rowKey":{"column":"line","value":7},"timestampColumn":"ts","type":"Float64"}
+            """;
+
+        SqlEquipmentTagParser.TryParse(json, "raw/1", out var def).ShouldBeTrue();
+
+        def.Model.ShouldBe(SqlTagModel.Query);
+        def.Name.ShouldBe("raw/1");
+        def.QueryName.ShouldBe("lineStatus");
+        def.ColumnName.ShouldBe("speed");
+        def.RowKeyColumn.ShouldBe("line");
+        def.RowKeyValue.ShouldBe("7");   // a JSON number is captured as text, like keyValue
+        def.TimestampColumn.ShouldBe("ts");
+        def.DeclaredType.ShouldBe(DriverDataType.Float64);
+        def.Table.ShouldBeEmpty();      // a Query tag reads no table of its own
+        SqlTagDefinitionFactory.Inspect(json).ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void TryParse_aQueryTagWithoutRowKey_isAccepted()
+        => SqlEquipmentTagParser.TryParse(
+            """{"model":"Query","query":"q","column":"c"}""", "raw", out _).ShouldBeTrue();
+
+    [Theory]
+    [InlineData("""{"model":"Query","column":"c"}""", "'query'")]
+    [InlineData("""{"model":"Query","query":"q"}""", "'column'")]
+    [InlineData("""{"model":"Query","query":"q","column":"c","rowKey":{"column":"k"}}""", "rowKey")]
+    [InlineData("""{"model":"Query","query":"q","column":"c","rowKey":{"value":"v"}}""", "rowKey")]
+    [InlineData("""{"model":"Query","query":"q","column":"c","rowKey":"k"}""", "rowKey")]
+    public void TryParse_anIncompleteQueryTag_isRejected_andTheGateSaysWhy(string json, string named)
+    {
+        SqlEquipmentTagParser.TryParse(json, "raw", out _).ShouldBeFalse();
+        SqlTagDefinitionFactory.Inspect(json).ShouldContain(w => w.Contains(named, StringComparison.Ordinal));
+    }
+
+    // ---- the read-only statement guard ----
+
+    [Theory]
+    [InlineData("SELECT a, b FROM t WHERE x = @x")]
+    [InlineData("  select a from t;  ")]
+    [InlineData("WITH latest AS (SELECT a FROM t) SELECT a FROM latest")]
+    [InlineData("SELECT 'DELETE FROM t; DROP TABLE t' AS note FROM t")]          // keywords inside a literal
+    [InlineData("SELECT a /* UPDATE t */ FROM t -- INSERT INTO t\n WHERE 1 = 1")]  // …and inside comments
+    [InlineData("SELECT [update] FROM t")]                                      // a quoted identifier
+    [InlineData("SELECT \"delete\" FROM t")]
+    public void ReadOnlyStatement_acceptsASingleRead(string sql)
+        => SqlReadOnlyStatement.Check(sql, out _).ShouldBeNull();
+
+    [Theory]
+    [InlineData("UPDATE t SET a = 1")]
+    [InlineData("DELETE FROM t")]
+    [InlineData("INSERT INTO t VALUES (1)")]
+    [InlineData("EXEC dbo.DoThings")]
+    [InlineData("SELECT a INTO t2 FROM t")]                                      // SELECT … INTO creates a table
+    [InlineData("SELECT a FROM t FOR UPDATE")]                                   // takes write locks
+    [InlineData("WITH d AS (DELETE FROM t RETURNING a) SELECT a FROM d")]         // data-modifying CTE
+    [InlineData("SELECT 1; DROP TABLE t")]                                       // a second statement
+    [InlineData("SELECT 1; SELECT 2")]
+    [InlineData("SELECT 'unterminated FROM t")]
+    [InlineData("SELECT 1 /* never closed")]
+    [InlineData("   ")]
+    public void ReadOnlyStatement_refusesAnythingElse(string sql)
+        => SqlReadOnlyStatement.Check(sql, out _).ShouldNotBeNull();
+
+    [Fact]
+    public void ReadOnlyStatement_stripsOneTrailingTerminator()
+    {
+        SqlReadOnlyStatement.Check("  SELECT a FROM t ;  ", out var normalised).ShouldBeNull();
+        normalised.ShouldBe("SELECT a FROM t");
+    }
+
+    [Theory]
+    [InlineData("SELECT a FROM t WHERE x = @line", "@line", true)]
+    [InlineData("SELECT a FROM t WHERE x = @lineId", "@line", false)]          // a longer name is not the marker
+    [InlineData("SELECT '@line' FROM t", "@line", false)]                       // inside a literal
+    [InlineData("SELECT a FROM t WHERE x = :p_line", ":p_line", true)]
+    public void UsesMarker_findsOnlyARealParameterToken(string sql, string marker, bool expected)
+        => SqlReadOnlyStatement.UsesMarker(sql, marker).ShouldBe(expected);
+
+    // ---- factory: named queries are validated, and one bad entry fails the driver ----
+
+    [Fact]
+    public void BuildOptions_typesAValidNamedQuery()
+    {
+        var options = SqlDriverFactoryExtensions.BuildOptions(
+            Dto("""{"q":{"sql":"SELECT a FROM t WHERE x = @x AND y = @y AND z = @z;","params":{"@x":7,"y":"s","z":1.5}}}"""),
+            "d1", new SqlServerDialect());
+
+        var query = options.Queries["q"];
+        query.Sql.ShouldBe("SELECT a FROM t WHERE x = @x AND y = @y AND z = @z");
+        query.ParameterNames.ShouldBe(["x", "y", "z"]);
+        query.ParameterValues.ShouldBe([7L, "s", 1.5]);
+        query.MaxRows.ShouldBe(SqlNamedQueryRules.DefaultMaxRows);
+    }
+
+    [Theory]
+    [InlineData("""{"q":{"sql":"DELETE FROM t"}}""", "single SELECT")]
+    [InlineData("""{"q":{"sql":"SELECT a FROM t","params":{"x":1}}}""", "'@x'")]       // declared, never used
+    [InlineData("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x y":1}}}""", "plain identifier")]
+    [InlineData("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x":[1]}}}""", "string, number")]
+    [InlineData("""{"q":{"sql":"SELECT a FROM t","maxRows":0}}""", "maxRows")]
+    public void BuildOptions_anUnusableNamedQuery_failsTheDriver(string queries, string named)
+    {
+        var ex = Should.Throw<InvalidOperationException>(() =>
+            SqlDriverFactoryExtensions.BuildOptions(Dto(queries), "d1", new SqlServerDialect()));
+        ex.Message.ShouldContain("d1");
+        ex.Message.ShouldContain(named);
+    }
+
+    [Fact]
+    public void BuildOptions_anOracleQueryMustUseTheOracleMarker()
+    {
+        Should.Throw<InvalidOperationException>(() => SqlDriverFactoryExtensions.BuildOptions(
+            Dto("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x":1}}}"""), "d1", new OracleDialect()));
+
+        SqlDriverFactoryExtensions.BuildOptions(
+                Dto("""{"q":{"sql":"SELECT a FROM t WHERE x = :p_x","params":{"x":1}}}"""), "d1", new OracleDialect())
+            .Queries["q"].ParameterNames.ShouldBe(["x"]);
+    }
+
+    // ---- planner ----
+
+    [Fact]
+    public void Plan_foldsEveryTagOfOneQueryIntoOnePlanRunningTheQueryVerbatim()
+    {
+        var queries = Queries(new SqlNamedQuery("q", "SELECT a, b FROM t WHERE x = @x", ["x"], [7L], 50));
+
+        var plan = SqlGroupPlanner.Plan(
+        [
+            QueryTag("A", "q", "a"),
+            QueryTag("B", "q", "b", rowKeyColumn: "a", rowKeyValue: "1"),
+        ], new SqlServerDialect(), queries).ShouldHaveSingleItem();
+
+        plan.Model.ShouldBe(SqlTagModel.Query);
+        plan.SqlText.ShouldBe("SELECT a, b FROM t WHERE x = @x");
+        plan.ParameterNames.ShouldBe(["@x"]);
+        plan.ParameterBindNames.ShouldBe(["@x"]);
+        plan.Parameters.ShouldBe([7L]);
+        plan.Members.Count.ShouldBe(2);
+        plan.RowCap.ShouldBe(50);
+        plan.SelectedColumns.ShouldBeEmpty();
+    }
+
+    [Fact]
+    public void Plan_aQueryTagNamingAnUndeclaredQuery_throws()
+        => Should.Throw<ArgumentException>(() =>
+            SqlGroupPlanner.Plan([QueryTag("A", "missing", "a")], new SqlServerDialect(), Queries()));
+
+    [Fact]
+    public void CatalogGate_passesQueryTagsThroughUntouched()
+    {
+        var tag = QueryTag("A", "q", "a");
+        var result = SqlCatalogGate.Apply([tag], SqlCatalog.Empty, new SqlServerDialect());
+        result.Accepted.ShouldHaveSingleItem().ShouldBeSameAs(tag);
+        result.Rejected.ShouldBeEmpty();
+    }
+
+    // ---- end to end over SQLite ----
+
+    [Fact]
+    public async Task Read_aRowKeyedQueryTag_readsItsRow_withBoundParameters()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture,
+            """
+            {"stations":{"sql":"SELECT station_id AS Station, oven_temp AS Temp, sample_ts AS Ts FROM LatestStatus WHERE pressure > @minPressure ORDER BY station_id",
+                         "params":{"minPressure":1.3}}}
+            """,
+            Tag("Q/Newest", """{"model":"Query","query":"stations","column":"temp","rowKey":{"column":"station","value":8},"timestampColumn":"ts","type":"Float64"}"""),
+            Tag("Q/Present", """{"model":"Query","query":"stations","column":"Temp","rowKey":{"column":"Station","value":"7"},"type":"Float64"}"""));
+
+        var snapshots = await driver.ReadAsync(["Q/Newest", "Q/Present"], CancellationToken.None);
+
+        // The parameter filters station 7 out (pressure 1.2): proof @minPressure was bound, not ignored.
+        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
+        Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture)
+            .ShouldBe(SqlitePollFixture.NewestStationOvenTemp);
+        snapshots[0].SourceTimestampUtc.ShouldBe(new DateTime(2026, 7, 24, 10, 0, 2, DateTimeKind.Utc));
+        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
+    }
+
+    [Fact]
+    public async Task Read_aSingleRowQuery_withoutRowKey_readsTheRow_andMoreRowsAreAmbiguous()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture,
+            """
+            {"one":{"sql":"SELECT MAX(num_value) AS top FROM TagValues"},
+             "many":{"sql":"SELECT num_value FROM TagValues"}}
+            """,
+            Tag("Q/One", """{"model":"Query","query":"one","column":"top","type":"Float64"}"""),
+            Tag("Q/Many", """{"model":"Query","query":"many","column":"num_value","type":"Float64"}"""));
+
+        var snapshots = await driver.ReadAsync(["Q/One", "Q/Many"], CancellationToken.None);
+
+        snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
+        Convert.ToDouble(snapshots[0].Value, CultureInfo.InvariantCulture).ShouldBe(SqlitePollFixture.PresentValue);
+        // Three rows and no rowKey: never "last row wins" — the selector is ambiguous.
+        snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadConfigurationError);
+        // …and that refusal is the database answering, so the driver stays Healthy.
+        driver.GetHealth().State.ShouldBe(DriverState.Healthy);
+    }
+
+    [Fact]
+    public async Task Read_aColumnTheQueryDoesNotReturn_isAConfigurationError_notSomeoneElsesCell()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture,
+            """{"one":{"sql":"SELECT MAX(num_value) AS top FROM TagValues"}}""",
+            Tag("Q/Wrong", """{"model":"Query","query":"one","column":"nope"}"""));
+
+        (await driver.ReadAsync(["Q/Wrong"], CancellationToken.None))[0].StatusCode
+            .ShouldBe(SqlStatusCodes.BadConfigurationError);
+    }
+
+    [Fact]
+    public async Task Read_moreRowsThanMaxRows_refusesTheWholeQuery()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture,
+            """{"all":{"sql":"SELECT tag_name, num_value FROM TagValues","maxRows":2}}""",
+            Tag("Q/Speed", $$$"""{"model":"Query","query":"all","column":"num_value","rowKey":{"column":"tag_name","value":"{{{SqlitePollFixture.PresentKey}}}"}}"""));
+
+        (await driver.ReadAsync(["Q/Speed"], CancellationToken.None))[0].StatusCode
+            .ShouldBe(SqlStatusCodes.BadConfigurationError);
+    }
+
+    [Fact]
+    public async Task Initialize_aQueryTagNamingAnUndeclaredQuery_isSkippedWithTheReason()
+    {
+        using var fixture = new SqlitePollFixture();
+        await using var driver = await StartAsync(fixture,
+            """{"one":{"sql":"SELECT 1 AS v"}}""",
+            Tag("Q/Ok", """{"model":"Query","query":"one","column":"v"}"""),
+            Tag("Q/Missing", """{"model":"Query","query":"nope","column":"v"}"""));
+
+        var skipped = ((ISkippedTagReporter)driver).GetSkippedTags().ShouldNotBeNull().Sample.ShouldHaveSingleItem();
+        skipped.RawPath.ShouldBe("Q/Missing");
+        skipped.Reason.ShouldContain("'nope'");
+        (await driver.ReadAsync(["Q/Missing", "Q/Ok"], CancellationToken.None))
+            .Select(s => s.StatusCode).ShouldBe([SqlStatusCodes.BadNodeIdUnknown, SqlStatusCodes.Good]);
+    }
+
+    /// <summary>
+    ///     The backstop behind the authoring guard: a named query runs inside a transaction that is always
+    ///     rolled back. The probe connection performs a WRITE inside every transaction the driver opens —
+    ///     standing in for a side-effecting function the lexical guard cannot see — so the table staying empty
+    ///     after a read proves the work was rolled back, and the begin counter (the positive control) proves a
+    ///     transaction was opened at all. A reader that committed would leave one row per poll.
+    /// </summary>
+    [Fact]
+    public async Task Read_aNamedQuery_runsInsideATransactionThatIsRolledBack()
+    {
+        using var fixture = new SqlitePollFixture();
+        fixture.Execute("CREATE TABLE SideEffects (n INTEGER)");
+        var factory = new SideEffectFactory();
+        await using var driver = await StartAsync(fixture, factory,
+            """{"one":{"sql":"SELECT MAX(num_value) AS top FROM TagValues"}}""",
+            Tag("Q/One", """{"model":"Query","query":"one","column":"top"}"""));
+
+        (await driver.ReadAsync(["Q/One"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
+        (await driver.ReadAsync(["Q/One"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
+
+        factory.TransactionsBegun.ShouldBe(2);
+        using var count = fixture.Connection.CreateCommand();
+        count.CommandText = "SELECT COUNT(*) FROM SideEffects";
+        Convert.ToInt64(count.ExecuteScalar(), CultureInfo.InvariantCulture).ShouldBe(0L);
+    }
+
+    [Fact]
+    public async Task Read_keyValueTags_doNotOpenATransaction()
+    {
+        using var fixture = new SqlitePollFixture();
+        fixture.Execute("CREATE TABLE SideEffects (n INTEGER)");
+        var factory = new SideEffectFactory();
+        await using var driver = await StartAsync(fixture, factory, queriesJson: null,
+            Tag("Kv", $$"""{"model":"KeyValue","table":"TagValues","keyColumn":"tag_name","keyValue":"{{SqlitePollFixture.PresentKey}}","valueColumn":"num_value"}"""));
+
+        (await driver.ReadAsync(["Kv"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
+        factory.TransactionsBegun.ShouldBe(0);
+    }
+
+    // ---- helpers ----
+
+    private static SqlDriverConfigDto Dto(string queriesJson) => System.Text.Json.JsonSerializer.Deserialize<SqlDriverConfigDto>(
+        new JsonObject { ["connectionStringRef"] = "x", ["queries"] = JsonNode.Parse(queriesJson) }.ToJsonString(),
+        SqlDriverFactoryExtensions.JsonOptionsForTest)!;
+
+    private static Dictionary<string, SqlNamedQuery> Queries(params SqlNamedQuery[] queries)
+        => queries.ToDictionary(q => q.Name, StringComparer.Ordinal);
+
+    private static SqlTagDefinition QueryTag(
+        string name, string query, string column, string? rowKeyColumn = null, string? rowKeyValue = null)
+        => new(name, SqlTagModel.Query, string.Empty, ColumnName: column, QueryName: query,
+            RowKeyColumn: rowKeyColumn, RowKeyValue: rowKeyValue);
+
+    private static RawTagEntry Tag(string rawPath, string json) => new(rawPath, json, WriteIdempotent: false);
+
+    private static Task<SqlDriver> StartAsync(SqlitePollFixture fixture, string? queriesJson, params RawTagEntry[] tags)
+        => StartAsync(fixture, fixture.Factory, queriesJson, tags);
+
+    private static async Task<SqlDriver> StartAsync(
+        SqlitePollFixture fixture, DbProviderFactory factory, string? queriesJson, params RawTagEntry[] tags)
+    {
+        var dto = queriesJson is null ? new SqlDriverConfigDto() : Dto(queriesJson);
+        var built = SqlDriverFactoryExtensions.BuildOptions(dto, DriverInstanceId, new SqliteDialect());
+        var options = new SqlDriverOptions
+        {
+            RawTags = tags,
+            OperationTimeout = TimeSpan.FromSeconds(15),
+            CommandTimeout = TimeSpan.FromSeconds(10),
+            Queries = built.Queries,
+        };
+        var driver = new SqlDriver(options, DriverInstanceId, new SqliteDialect(), fixture.ConnectionString, factory);
+        await driver.InitializeAsync("{}", CancellationToken.None);
+        return driver;
+    }
+
+    /// <summary>A SQLite factory whose connections write a row inside every transaction they begin.</summary>
+    private sealed class SideEffectFactory : DbProviderFactory
+    {
+        private int _begun;
+
+        public int TransactionsBegun => Volatile.Read(ref _begun);
+
+        public override DbConnection CreateConnection() => new SideEffectConnection(this);
+
+        private sealed class SideEffectConnection(SideEffectFactory owner) : SqliteConnection
+        {
+            protected override DbTransaction BeginDbTransaction(IsolationLevel isolationLevel)
+            {
+                Interlocked.Increment(ref owner._begun);
+                var transaction = BeginTransaction(isolationLevel);
+                using var write = CreateCommand();
+                write.Transaction = transaction;
+                write.CommandText = "INSERT INTO SideEffects (n) VALUES (1)";
+                write.ExecuteNonQuery();
+                return transaction;
+            }
+        }
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/SqlDriverFormContractTests.cs

@@ -130,4 +130,58 @@ public sealed class SqlDriverFormContractTests
 
         json.ShouldContain($"\"provider\":\"{reserved}\"");
     }
+    // ---- named queries (lmxopcua#753): the form refuses exactly what the factory refuses ----
+
+    /// <summary>
+    ///     The driver form's queries editor consults the runtime's own rules
+    ///     (<see cref="SqlNamedQueryRules.Describe"/>), so what the form keeps is what the factory constructs and
+    ///     what it refuses the factory refuses too. Driven through the real factory entry point, so a rule added
+    ///     on one side only breaks this test.
+    /// </summary>
+    [Theory]
+    [InlineData("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x":1}}}""", true)]
+    [InlineData("""{"q":{"sql":"WITH l AS (SELECT a FROM t) SELECT a FROM l;","maxRows":10}}""", true)]
+    [InlineData("""{"q":{"sql":"UPDATE t SET a = 1"}}""", false)]
+    [InlineData("""{"q":{"sql":"SELECT a INTO t2 FROM t"}}""", false)]
+    [InlineData("""{"q":{"sql":"SELECT 1; SELECT 2"}}""", false)]
+    [InlineData("""{"q":{"sql":"SELECT a FROM t","maxRows":0}}""", false)]
+    [InlineData("""{"q":{"sql":"SELECT a FROM t WHERE x = @x","params":{"x":{"nested":1}}}}""", false)]
+    [InlineData("""{"q":{"sql":""}}""", false)]
+    public void Form_queries_editor_and_factory_agree(string queriesJson, bool accepted)
+    {
+        var formError = SqlDriverForm.FormModel.TryParseQueries(queriesJson, out _);
+        (formError is null).ShouldBe(accepted, formError);
+
+        var envVar = SqlConnectionStringResolver.EnvironmentVariableFor("AdminUiFormTest");
+        var previous = Environment.GetEnvironmentVariable(envVar);
+        Environment.SetEnvironmentVariable(envVar, "Server=localhost;Database=Test;Trusted_Connection=True;");
+        try
+        {
+            var blob = $$"""{"provider":"SqlServer","connectionStringRef":"AdminUiFormTest","queries":{{queriesJson}}}""";
+            if (accepted) { SqlDriverFactoryExtensions.CreateInstance("adminui-form-test", blob).ShouldNotBeNull(); }
+            else
+            {
+                Should.Throw<InvalidOperationException>(
+                    () => SqlDriverFactoryExtensions.CreateInstance("adminui-form-test", blob));
+            }
+        }
+        finally
+        {
+            Environment.SetEnvironmentVariable(envVar, previous);
+        }
+    }
+
+    [Fact]
+    public void Queries_survive_the_form_model_round_trip_and_a_typo_keeps_the_last_valid_set()
+    {
+        var dto = JsonSerializer.Deserialize<SqlDriverConfigDto>(
+            """{"provider":"SqlServer","connectionStringRef":"R","queries":{"q":{"sql":"SELECT a FROM t"}}}""", FormOptions)!;
+        var form = SqlDriverForm.FormModel.FromDto(dto);
+
+        form.SetQueriesJson("{ not json");
+        form.QueriesError.ShouldNotBeNull();
+
+        var json = JsonSerializer.Serialize(form.ToDto(), FormOptions);
+        json.ShouldContain("\"queries\":{\"q\":{\"sql\":\"SELECT a FROM t\"}}");
+    }
 }

tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/SqlTagConfigModelTests.cs

@@ -232,10 +232,42 @@ public sealed class SqlTagConfigModelTests
             """{"model":"KeyValue","table":"t","keyColumn":"k","keyValue":"v","valueColumn":"c","type":"Bogus"}""")
             .Validate().ShouldNotBeNull();
 
+    [Theory]
+    [InlineData("""{"model":"Query","table":"t"}""")]                                                   // no query, no column
+    [InlineData("""{"model":"Query","query":"q"}""")]                                                   // no column
+    [InlineData("""{"model":"Query","query":"q","column":"c","rowKey":{"column":"k"}}""")]             // half a rowKey
+    [InlineData("""{"model":"Query","query":"q","column":"c","rowKey":"k"}""")]                        // not an object
+    public void Validate_rejects_an_incomplete_query_tag(string json)
+        => SqlTagConfigModel.FromJson(json).Validate().ShouldNotBeNull();
+
     [Fact]
-    public void Validate_rejects_deferred_query_model()
-        => SqlTagConfigModel.FromJson("""{"model":"Query","table":"t"}""")
-            .Validate().ShouldNotBeNull();
+    public void Query_tag_round_trips_and_parses_to_the_expected_definition()
+    {
+        var json = SqlTagConfigModel.FromJson(
+            """{"model":"Query","query":"lineStatus","column":"speed","rowKey":{"column":"line","value":7,"note":"keep"},"timestampColumn":"ts","type":"Float64"}""")
+            .ToJson();
+
+        SqlTagConfigModel.FromJson(json).Validate().ShouldBeNull();
+        json.ShouldContain("\"note\":\"keep\"");   // unknown rowKey keys survive an edit
+        SqlEquipmentTagParser.TryParse(json, "raw/path", out var def).ShouldBeTrue();
+        def.Model.ShouldBe(SqlTagModel.Query);
+        def.QueryName.ShouldBe("lineStatus");
+        def.ColumnName.ShouldBe("speed");
+        def.RowKeyColumn.ShouldBe("line");
+        def.RowKeyValue.ShouldBe("7");
+        def.TimestampColumn.ShouldBe("ts");
+    }
+
+    [Fact]
+    public void Clearing_the_row_key_omits_it_from_the_blob()
+    {
+        var model = SqlTagConfigModel.FromJson("""{"model":"Query","query":"q","column":"c","rowKey":{"column":"k","value":"v"}}""");
+        model.RowKey.Clear();
+        var json = model.ToJson();
+        json.ShouldNotContain("rowKey");
+        SqlEquipmentTagParser.TryParse(json, "raw", out var def).ShouldBeTrue();
+        def.RowKeyColumn.ShouldBeNull();
+    }
 
     // ---- LOAD-BEARING: editor output MUST parse through the runtime factory reader --------------
 

tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/TagConfigAuthoringDeployParityTests.cs

@@ -68,6 +68,20 @@ public sealed class TagConfigAuthoringDeployParityTests
         AssertParity(DriverTypeNames.Sql, model.Validate(), model.ToJson());
     }
 
+    /// <summary>A named-query Sql tag (lmxopcua#753) as the typed editor saves it.</summary>
+    [Fact]
+    public void Sql_query_config_the_editor_accepts_draws_no_deploy_warning()
+    {
+        var model = new SqlTagConfigModel
+        {
+            Model = ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts.SqlTagModel.Query,
+            QueryName = "lineStatus",
+            Column = "speed",
+            RowKey = new SqlRowKeyModel { Column = "line", Value = "Line1" },
+        };
+        AssertParity(DriverTypeNames.Sql, model.Validate(), model.ToJson());
+    }
+
     [Fact]
     public void MTConnect_config_the_editor_accepts_draws_no_deploy_warning()
     {

tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/TagConfigNonEnumParityTests.cs

@@ -403,6 +403,13 @@ public sealed class TagConfigNonEnumParityTests
         (DriverTypeNames.Sql, """{"model":"KeyValue","table":"dbo.T","keyColumn":"k","valueColumn":"v","keyValue":"x"}"""),
         (DriverTypeNames.Sql, """{"table":"dbo.T","keyColumn":"k","valueColumn":"v","keyValue":"x"}"""),
         (DriverTypeNames.Sql, """{"model":"Query","table":"dbo.T"}"""),
+        // lmxopcua#753 — the named-query model: query + column required, rowKey all-or-nothing.
+        (DriverTypeNames.Sql, """{"model":"Query","query":"lineStatus","column":"speed"}"""),
+        (DriverTypeNames.Sql, """{"model":"Query","query":"lineStatus","column":"speed","rowKey":{"column":"line","value":7}}"""),
+        (DriverTypeNames.Sql, """{"model":"Query","query":"lineStatus","column":"speed","rowKey":{"column":"line"}}"""),
+        (DriverTypeNames.Sql, """{"model":"Query","query":"lineStatus","column":"speed","rowKey":"line"}"""),
+        (DriverTypeNames.Sql, """{"model":"Query","query":"lineStatus"}"""),
+        (DriverTypeNames.Sql, """{"model":"Query","column":"speed","query":"   "}"""),
         (DriverTypeNames.Sql, """{"model":"WideRow","table":"dbo.T","columnName":"Speed","rowSelector":{"whereColumn":"LineId","whereValue":"L1"}}"""),
         (DriverTypeNames.Sql, """{"model":"WideRow","table":"dbo.T","columnName":"Speed"}"""),
 
@@ -611,6 +618,13 @@ public sealed class TagConfigNonEnumParityTests
             ColumnName = "Speed",
             RowSelector = new SqlRowSelectorModel { WhereColumn = "LineId", WhereValue = "L1" },
         }.ToJson());
+        data.Add(DriverTypeNames.Sql, new SqlTagConfigModel
+        {
+            Model = SqlTagModel.Query,
+            QueryName = "lineStatus",
+            Column = "speed",
+            RowKey = new SqlRowKeyModel { Column = "line", Value = "Line1" },
+        }.ToJson());
         return data;
     }