feat(sql): authored write block — Update/Insert/Upsert/Procedure + timestamp stamp; ODBC scope + login docs (lmxopcua#781)
v2-ci / unit-tests (push) Has been cancelled

Item 2: a tag's optional `write` block picks the statement the driver builds from its own
catalog-validated identifiers (no authored SQL; values always bound):
- Update (default, unchanged back-compat UPDATE ... WHERE key; 0 rows = BadNoData)
- Insert (key+value[+stamp]; append for a WideRow topByTimestamp tag)
- Upsert: SQL Server MERGE WITH (HOLDLOCK), PostgreSQL INSERT ... ON CONFLICT, MySQL
  ON DUPLICATE KEY UPDATE (1-or-2 rows), Oracle MERGE from DUAL
- Procedure: named stored procedure resolved against the routine catalog by the catalog gate
  (INFORMATION_SCHEMA.ROUTINES / ALL_PROCEDURES), value/key/timestamp params by name (MySQL positional)
- stampTimestamp sets the tag's timestamp column to the write time (UTC)
Odbc serves Update/Insert only (shared SqlWriteSupport table). Constraint failures map to
BadEntryExists (unique) / BadOutOfRange (other integrity, data exception); everything else,
permission denied included, stays BadCommunicationError. WriteIdempotent stays false.
The parser refuses malformed blocks (strict keys/enums/param charset); DraftValidator emits
blocking SqlWriteBlockInvalid / SqlWriteModeUnsupported from the driver's own parse; the AdminUI
editor surfaces the same message.

Items 1 + 4 (docs/drivers/Sql.md): ODBC stays the ANSI SQL:2008 profile (native providers for
SQL Server/MySQL, DB2 unsupported) with an ops note on driver manager + vendor driver on every
Odbc and AdminUI host; read-only drivers require a SELECT-only login, writable drivers a login
limited to their write targets.

Tests: SqlWriteModeTests (parse, every dialect's SQL, gate, status map, SQLite end-to-end),
DraftValidatorSqlWriteBlockTests, AdminUI round-trip; live SqlWriteModesLiveTests on SQL Server,
PostgreSQL, MySQL, Oracle (seeds gain write_targets + set_target) and an Odbc write leg.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-09 22:56:21 -05:00
parent 2da58d0360
commit f5c30ed0ef
33 changed files with 2361 additions and 132 deletions
+1 -1
View File
@@ -73,7 +73,7 @@ Three properties worth knowing:
| [OPC UA Client](OpcUaClient.md) | `Driver.OpcUaClient` | B | OPCFoundation `Opc.Ua.Client` | IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IAlarmSource, IHostConnectivityProbe | Gateway/aggregation driver — does not proxy upstream history (no driver has a history capability). Opens a single `Session` against a remote OPC UA server and re-exposes its address space. Owns its own `ApplicationConfiguration` (distinct from `Client.Shared`) because it's always-on with keep-alive + `TransferSubscriptions` across SDK reconnect, not an interactive CLI |
| [MQTT](Mqtt.md) | `Driver.Mqtt` (+ `.Browser`, `.Contracts`) | A | [MQTTnet 5](https://github.com/dotnet/MQTTnet) | IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, IRediscoverable | **Push, not poll, and read-only** — the broker delivers PUBLISH frames the driver forwards straight to `ISubscribable.OnDataChange`; `IReadable` serves the last retained/received value from cache, and there is **no `IWritable`** (nothing publishes back). Subscriptions are indexed **by topic filter, not by topic**, so wildcard tags survive a reconnect. A rejected CONNACK throws `MqttConnectRejectedException`: unrecoverable codes → `Faulted` + supervisor stop, transient → retry. **Two modes:** `Plain` binds a tag to a concrete topic + JSON path; `SparkplugB` decodes vendored Eclipse-Tahu protobuf under one `spBv1.0/{groupId}/#` subscription and binds by the `group/edgeNode/device?/metric` tuple (birth/alias/seq-gap state machine, death→STALE, scoped rebirth NCMD). `IRediscoverable` fires on a DBIRTH but is **inert platform-side** — see [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) |
| [MTConnect](MTConnect.md) | `Driver.MTConnect` (+ `.Contracts`) | — | Hand-rolled `System.Xml.Linq` HTTP/XML client against a vendor-neutral MTConnect Agent (`/probe`, `/current`, `/sample`) — no TrakHound dependency (dropped; see the driver doc) | IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, IRediscoverable | Read-only by design (no `IWritable`). Production data plane is entirely `ISubscribable`/`OnDataChange` — `IReadable` is CLI/test-only. Browse is free via the Wave-0 universal discovery browser, no bespoke browser project. `IRediscoverable`/`IHostConnectivityProbe` have no server-side consumer yet — a pre-existing fleet-wide gap, not MTConnect-specific |
| [Sql](Sql.md) | `Driver.Sql` (+ `.Browser`, `.Contracts`) | A | `Microsoft.Data.SqlClient` behind an `ISqlDialect` seam (`SqlServerDialect` shipped; Postgres/ODBC deferred) | IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe | Read-only DB-poll driver with a bespoke schema browser. **Credentials are never persisted** — the deploy gate rejects a stored `connectionString` (#498) and catalog identifiers are validated against the live catalog's own spelling (#496). `SupportsOnlineDiscovery => false` by design (the bespoke browser wins). `SqlTagModel.Query` is deferred to P3 and rejected end-to-end by parser, validator, planner and editor. Merged `4ad54037` |
| [Sql](Sql.md) | `Driver.Sql` (+ `.Browser`, `.Contracts`, `.Dialect`) | A | ADO.NET behind an `ISqlDialect` seam — `SqlServer` (`Microsoft.Data.SqlClient`), `Postgres` (Npgsql), `Oracle` (ODP.NET), `MySql` (MySqlConnector), `Odbc` (ANSI SQL:2008 profile only) | IDriver, ITagDiscovery, IReadable, IWritable (opt-in), ISubscribable, IHostConnectivityProbe | DB-poll driver (KeyValue / WideRow / named read-only `Query` tags) with a bespoke schema browser. **Credentials are never persisted** — the deploy gate rejects a stored `connectionString` (#498) and catalog identifiers are validated against the live catalog's own spelling (#496). Writes are off unless `allowWrites` + the tag's `writable`; a tag's `write` block picks Update / Insert / Upsert / stored Procedure (#781; Odbc: Update/Insert). `SupportsOnlineDiscovery => false` by design (the bespoke browser wins). |
| [Calculation](Calculation.md) | `Driver.Calculation` | A | none — computes from other tags' RawPaths | IDriver, IDependencyConsumer, ISubscribable, IReadable | **Pseudo-driver**, not a wire protocol: signal-level tags computed from other tags via a script, with deploy-time `scriptId`-existence + Tarjan cycle gates. Its probe is parse-only (no backend to reach). `runTimeoutMs` is authorable through `CalculationDriverForm` (the "no typed config form" gap — G-1 — was closed 2026-07-27) |
| [Historian.Gateway](../Historian.md) | `Driver.Historian.Gateway` | — | `ZB.MOM.WW.HistorianGateway.Client` gRPC (`historian_gateway.v1`) | IHistorianDataSource (server-side read backend) + alarm `SendEvent` writer + `WriteLiveValues` recorder + `IHistorianProvisioning` | Not a tag driver — the default historian backend (the other is `Historian.SqlServer`, below). Registers `GatewayHistorianDataSource : IHistorianDataSource` for HistoryRead and serves alarm-write + continuous historization through the gateway. No `IDriver`/`ITagDiscovery` surface. (The retired Wonderware sidecar backend it replaced is documented at [Historian.Wonderware.md](Historian.Wonderware.md).) |
| [Historian.SqlServer](../Historian.md#sql-server-backend) | `Driver.Historian.SqlServer` | — | `Microsoft.Data.SqlClient` (a plain SQL Server database, schema `hist` by default) | `IHistorianDataSource` + `IHistorianValueWriter` + `IAlarmHistorianWriter` + `IHistorianProvisioning` + `IHistorianDataUpdater` | Not a tag driver — a pluggable historian backend selected with `ServerHistorian:Backend=SqlServer`. Stores typed values with their real status, serves Raw / AtTime / Events / 13 push-down aggregates, and is the only backend that supports OPC UA **HistoryUpdate**. No `IDriver`/`ITagDiscovery` surface. |
+145 -42
View File
@@ -5,7 +5,7 @@
Getting-started guide for the `Sql` driver — an Equipment-kind driver that polls rows out of a
relational database and publishes them as OPC UA variables. It is **read-only unless writes are
explicitly enabled** at both the driver and the tag (see [Writes](#writes-opt-in-753)). For the design rationale read
explicitly enabled** at both the driver and the tag (see [Writes](#writes-opt-in-753-781)). For the design rationale read
[`docs/plans/2026-07-15-sql-poll-driver-design.md`](../plans/2026-07-15-sql-poll-driver-design.md);
for the build-vs-plan record read
[`docs/plans/2026-07-24-sql-poll-driver.md`](../plans/2026-07-24-sql-poll-driver.md).
@@ -32,19 +32,40 @@ boolean under MySqlConnector's default `TreatTinyAsBoolean`; `BIT(n)`, `TIME`, `
types infer as `String`. Live gate: `Docker/mysql/docker-compose.yml` + `MySqlReadTests`
(`SQL_MYSQL_ENDPOINT` / `SQL_MYSQL_PASSWORD`).
ODBC specifics: `OdbcDialect` is the **ANSI SQL:2008 profile** — double-quoted identifiers,
positional `?` parameters, the standard `INFORMATION_SCHEMA` views and `FETCH FIRST n ROWS ONLY`.
ODBC abstracts the wire, not the SQL, so the backend behind the DSN must accept all four:
PostgreSQL (psqlODBC — the live gate), Snowflake and similar do; **SQL Server** (needs `OFFSET …
FETCH` after an `ORDER BY`), **MySQL** (no `FETCH FIRST`) and **DB2** (no `INFORMATION_SCHEMA`) do
not and should use their native provider. It is also the one provider that is **not fully
managed**: every host running an Odbc instance needs the platform driver manager (unixODBC / iODBC
/ Windows' own) and the vendor's ODBC driver — on the driver nodes and, for browse, the AdminUI
nodes. Parameters bind in order, so the planner, the catalog loader and the browse session all add
them in marker order, and a named query's `params` bind in authored order (the factory refuses a
statement whose `?` count differs). Live gate: `Docker/odbc/run-odbc-gate.sh` runs `OdbcReadTests`
in a throwaway Linux container with unixODBC + psqlODBC against a throwaway PostgreSQL; on macOS
the suite is unit-tested only (`OdbcDialectTests`) because no vendor ODBC driver is installed.
ODBC specifics: `OdbcDialect` is the **ANSI SQL:2008 profile, and only that** — double-quoted
identifiers, positional `?` parameters, the standard `INFORMATION_SCHEMA` views and `FETCH FIRST n
ROWS ONLY`. The scope is deliberately narrow (owner decision 2026-11-09, lmxopcua#781): there are no
per-backend ODBC dialect variants and no `SQLTables`/`SQLColumns` catalog path. ODBC abstracts the
wire, not the SQL, so the backend behind the connection string must accept all four: PostgreSQL
(psqlODBC — the live gate), Snowflake and similar do. **SQL Server** (needs `OFFSET … FETCH` after an
`ORDER BY`) and **MySQL / MariaDB** (no `FETCH FIRST`) do not — use the native `SqlServer` / `MySql`
provider. **DB2** (no `INFORMATION_SCHEMA`) is not supported through Odbc at all. Writes over Odbc
are **Update and Insert only** — see [Writes](#writes-opt-in-753-781).
Parameters bind in order, so the planner, the catalog loader and the browse session all add them in
marker order, and a named query's `params` bind in authored order (the factory refuses a statement
whose `?` count differs). Live gate: `Docker/odbc/run-odbc-gate.sh` runs `OdbcReadTests` (reads,
named queries, browse, and Update/Insert writes) in a throwaway Linux container with unixODBC +
psqlODBC against a throwaway PostgreSQL; on macOS the suite is unit-tested only (`OdbcDialectTests`)
because no vendor ODBC driver is installed.
> **Ops note — Odbc is the one provider that is not fully managed.** `System.Data.Odbc` is only the
> ADO.NET shim; the work is done by native code that must be installed on **every host that runs an
> Odbc driver instance** (every node of its cluster, since either can become Primary) **and every
> AdminUI host** (the browse picker and Test Connect open the connection from the AdminUI process):
>
> - **Linux:** the driver manager (`unixodbc`) plus the vendor's driver package (e.g.
> `odbc-postgresql`), registered in `odbcinst.ini`. Check with `odbcinst -q -d` — the name it prints
> is the `Driver={…}` value the connection string needs.
> - **Windows:** the driver manager ships with the OS; install the vendor's **64-bit** ODBC driver
> (the server is x64) and confirm it in *ODBC Data Sources (64-bit)* → *Drivers*.
> - Prefer a **DSN-less** connection string (`Driver={PostgreSQL Unicode};Server=…;Database=…`) in
> the secret store: a DSN is per-host configuration that every node and AdminUI host would also
> need, identically.
>
> A host missing either piece fails the driver's Initialize (and the AdminUI browse) with the driver
> manager's "data source name not found / no default driver" error — a host-provisioning fault, not a
> tag fault.
Two provider-specific things are worth knowing before you author against them:
@@ -56,7 +77,7 @@ Two provider-specific things are worth knowing before you author against them:
(see [Optional per-tag fields](#optional-per-tag-fields)) when the OPC UA type matters.
**Read-only by default.** `allowWrites` is off unless authored `true`, and even then only a tag whose
own TagConfig says `"writable": true` can be written — see [Writes](#writes-opt-in-753).
own TagConfig says `"writable": true` can be written — see [Writes](#writes-opt-in-753-781).
## Tag models
@@ -173,7 +194,8 @@ row:
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.
procedure, a sequence, a call out to another system) is outside both — which is why a read-only
driver **must** connect with a `SELECT`-only login (see [Database login](#database-login-required)).
- 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.
@@ -187,6 +209,7 @@ row:
| `timestampColumn` | Source timestamp for the value. Absent ⇒ the driver stamps the poll time. |
| `type` | Explicit OPC UA type override. Absent ⇒ inferred from the result set's column metadata. A write is coerced to it. |
| `writable` | The tag's write opt-in — a JSON **boolean**; absent ⇒ read-only. Anything else (`"true"`, `1`) refuses the tag. Effective only with the driver's `allowWrites`. |
| `write` | Optional: how a write is executed — `Update` (default), `Insert`, `Upsert` or `Procedure`, plus the timestamp stamp. See [Writes](#writes-opt-in-753-781). A malformed block refuses the whole tag and blocks deploy. |
There is **no per-tag poll-interval field**. The interval is the subscriber's, floored by
`PollGroupEngine.DefaultMinInterval`, falling back to the instance's `defaultPollInterval` when the
@@ -216,7 +239,7 @@ existed in the parser — nothing reads it.)
| `commandTimeout` | `00:00:10` | ADO.NET server-side backstop. **Must be less than `operationTimeout`** — it is the backstop, not the deadline. |
| `maxConcurrentGroups` | `4` | Connection-pool guard: cap on group queries in flight. |
| `nullIsBad` | `false` | `false` ⇒ a NULL cell publishes **Uncertain**; `true` ⇒ **Bad**. |
| `allowWrites` | `false` | Driver-level write switch. Off ⇒ every write is `BadNotWritable`. See [Writes](#writes-opt-in-753). |
| `allowWrites` | `false` | Driver-level write switch. Off ⇒ every write is `BadNotWritable`. See [Writes](#writes-opt-in-753-781). |
| `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
@@ -329,42 +352,122 @@ catalog can supply it. `RawBrowseCommitMapper.DescribeUncommittableLeaf` refuses
names the table, and points at the tag editor; no selector is invented, and the mapper's blob stays
selector-less rather than dropping to a generic `{"address": …}` the typed editor cannot read.
## Writes (opt-in, #753)
## Writes (opt-in, #753, #781)
A write reaches the database only when **every** gate holds, outermost first:
1. the OPC UA node is Writable (the Tag's own flag) and the caller holds `WriteOperate` — the server;
2. the driver config says `"allowWrites": true` — otherwise `BadNotWritable`;
3. the tag's TagConfig says `"writable": true` — otherwise `BadNotWritable`;
4. the model names exactly one row: **KeyValue**, or **WideRow with a `whereColumn`/`whereValue`
pair**. `topByTimestamp` ("the newest row" moves under a concurrent insert) and `Query` (a projection
names no table) are `BadNotWritable`.
4. the tag's write mode has a target on its model (table below) — otherwise `BadNotWritable`;
5. the provider serves the mode — otherwise deploy refuses it (`SqlWriteModeUnsupported`) and, at
runtime, `BadNotSupported`.
The write is the **only** statement shape the driver can produce — there is no authored write SQL,
INSERT, UPSERT or procedure call:
There is **no authored write SQL.** A tag's optional `write` block picks one of four statement
shapes the driver builds itself from the tag's own catalog-validated identifiers:
```sql
UPDATE <table> SET <valueColumn | columnName> = @v WHERE <keyColumn | whereColumn> = @k
```json
{ "model": "KeyValue", "table": "plant.setpoints", "keyColumn": "tag_name", "keyValue": "Line1.Speed",
"valueColumn": "sp_value", "timestampColumn": "updated_at", "type": "Float64", "writable": true,
"write": { "mode": "Upsert", "stampTimestamp": true } }
```
- **Identifiers are the catalog gate's**: a write resolves against the gate-accepted definition, so
every name in that statement was read back out of the catalog and dialect-quoted. A gate-rejected tag
answers `BadNodeIdUnknown`. The value and the key are always **bound parameters**.
- **Exactly one row or nothing.** The UPDATE runs in a transaction that commits only when it touched
one row. Zero rows ⇒ `BadNoData` (the driver never inserts); more than one ⇒ rolled back,
`BadConfigurationError` (the selector is ambiguous — the read path's rule); a provider that cannot
report a row count ⇒ rolled back, `BadNotSupported`.
- The value is coerced to the tag's `type` when it has one (`BadTypeMismatch` when it will not
coerce), else bound as the scalar the client sent. A null writes SQL NULL.
- Bounded by `operationTimeout` (`BadTimeout`). A provider error after the connection opened — a
missing `UPDATE` grant, a constraint — is `BadCommunicationError`, nothing committed. Writes never
change driver health.
- Discovery advertises `Operate` only for a tag where all of 2–4 hold, `ViewOnly` otherwise.
- **Grant the login `UPDATE` on just the tables that carry writable tags.** The live gate
(`MySqlReadTests`) proves the server refuses a write to a SELECT-only table even with every driver
| `mode` | Statement | Models | Rows that count as success |
|---|---|---|---|
| `Update` *(default — also what a writable tag with no `write` block does)* | `UPDATE t SET value [, stamp] WHERE key = @k` | KeyValue; WideRow with `whereColumn`/`whereValue` | exactly 1 |
| `Insert` | `INSERT INTO t (key, value [, stamp]) VALUES (…)` — no key column for a WideRow `topByTimestamp` tag (an append) | KeyValue; WideRow (either selector) | exactly 1 |
| `Upsert` | SQL Server `MERGE … WITH (HOLDLOCK)`; PostgreSQL `INSERT … ON CONFLICT (key) DO UPDATE`; MySQL `INSERT … ON DUPLICATE KEY UPDATE`; Oracle `MERGE … USING (SELECT 1 FROM dual)` | KeyValue; WideRow with a where-pair | 1 (MySQL: 1 or 2) |
| `Procedure` | SQL Server `EXEC p @value = @v, …`; PostgreSQL `CALL p(value => @v, …)`; MySQL `CALL p(@v, …)` (**positional**); Oracle `BEGIN p(value => :p_v, …); END;` | any, including `Query` | the call did not throw |
`write` keys — anything else refuses the tag:
| Key | Modes | Meaning |
|---|---|---|
| `mode` | all | `Update` \| `Insert` \| `Upsert` \| `Procedure`; absent ⇒ `Update`. |
| `stampTimestamp` | Update / Insert / Upsert | `true` ⇒ also set the tag's `timestampColumn` (a WideRow `topByTimestamp` tag: that column) to the write time, **UTC, from the driver host's clock**. Requires such a column. Default `false` — the timestamp column is then untouched and the next read keeps the row's old timestamp unless the database maintains it (a trigger, `ON UPDATE CURRENT_TIMESTAMP`). |
| `procedure` | Procedure (required) | `procedure` or `schema.procedure` — a **stored procedure**, resolved against the routine catalog like a table (case folds; an unknown, non-executable or package-member name rejects the tag). |
| `valueParam` | Procedure | The procedure parameter the value binds to. Default `value`. |
| `keyParam` | Procedure | Optional: the parameter the tag's key binds to (`keyValue`, `whereValue` or a Query tag's `rowKey.value`). |
| `timestampParam` | Procedure | Optional: the parameter the write time (UTC) binds to. |
Parameter names are plain identifiers (letter/underscore, then letters/digits/underscores; a leading
`@`/`:` is stripped) — they reach the statement unquoted, in named notation. **MySQL has no named
notation**: its arguments are passed positionally in the fixed order *value, key, timestamp* (those
configured), so declare the procedure's parameters in that order. **PostgreSQL** resolves procedure
arguments by implicit cast, so declare a numeric value parameter `double precision` (a `Float64`
binds as that) and a timestamp parameter `timestamptz`.
**Provider support** (one table, `SqlWriteSupport`, read by the driver and the deploy gate):
| | Update | Insert | Upsert | Procedure |
|---|---|---|---|---|
| SqlServer, Postgres, MySql, Oracle | ✓ | ✓ | ✓ | ✓ |
| Odbc | ✓ | ✓ | — | — |
Odbc has no upsert every ODBC backend accepts (and positional `?` markers cannot repeat, which every
single-statement upsert needs) and no portable routine catalog or calling convention; use the
backend's native provider for those modes.
**Upsert needs a unique key on the key column.** PostgreSQL refuses `ON CONFLICT (key)` without a
unique constraint or index on exactly that column; MySQL's `ON DUPLICATE KEY` fires on *any* unique
key of the table, so the key column should be the primary key and no other unique key should be able
to collide; SQL Server and Oracle `MERGE` work without one, but then a key matching two rows updates
both and is rolled back (`BadConfigurationError`).
### Outcomes
| Outcome | Status |
|---|---|
| Committed | `Good` |
| `Update` (or a non-MySQL `Upsert`) touched no row — the key is absent; nothing is inserted | `BadNoData` |
| More rows than the mode allows — the key is not unique; rolled back | `BadConfigurationError` |
| The provider could not report a row count; rolled back | `BadNotSupported` |
| Unique / primary-key violation — e.g. an `Insert` of a key that exists (SQLSTATE 23505, SQL Server 2627/2601, MySQL 1062, ORA-00001) | `BadEntryExists` |
| Another integrity rule — NOT NULL, foreign key, CHECK (SQLSTATE class 23, SQL Server 515/547, ORA-01400/02290/02291) — or a data exception: overflow, truncation, an unconvertible value (SQLSTATE class 22, SQL Server 8115/8152/245, ORA-01438/12899/01722) | `BadOutOfRange` |
| Any other provider error after the connection opened — **a missing grant included**, a missing object, a broken link | `BadCommunicationError` |
| The value will not coerce to the tag's `type` | `BadTypeMismatch` |
| `operationTimeout` elapsed | `BadTimeout` |
Nothing is committed on any non-`Good` outcome. Writes are **never retried**
(`WriteIdempotent = false` — an `Insert` or a procedure is not idempotent, and an `Update` is only
idempotent until something else writes the row), and a failed write reverts the node to its prior
value. Writes never change driver health. Discovery advertises `Operate` only for a tag where gates
2–5 hold, `ViewOnly` otherwise.
### Safety
- **Identifiers are the catalog gate's.** A write resolves against the gate-accepted definition, so
every table, column and procedure name in the statement was read back out of the catalog
(`INFORMATION_SCHEMA.ROUTINES` / Oracle `ALL_PROCEDURES` for procedures) and dialect-quoted. A
gate-rejected tag answers `BadNodeIdUnknown`.
- **Values are always bound** — the written value, the key and the stamp. Nothing an OPC UA client
sends reaches command text.
- **Deploy checks the block with the driver's own parser** (`SqlTagDefinitionFactory.DescribeWriteBlock`):
a malformed block is the blocking error `SqlWriteBlockInvalid`, a mode the instance's provider does
not serve is `SqlWriteModeUnsupported`. The AdminUI tag editor shows the same parser message.
- Live gate: `SqlWriteModesLiveTests` runs the same suite (Update+stamp, absent key, Insert and its
duplicate, NOT NULL, Upsert insert-then-update, Procedure, unknown procedure) against SQL Server,
PostgreSQL, MySQL and Oracle; `OdbcReadTests` covers Odbc's Update/Insert.
## Database login (required)
The driver's own guards are not a substitute for the database's grants:
- **A read-only driver (`allowWrites` off) must connect with a `SELECT`-only login** — `SELECT` on
the tables and views its tags and named queries read, and nothing else. The named-query guard is a
keyword check plus an always-rolled-back transaction; neither can stop a **non-transactional** side
effect — a sequence `nextval`, a function that sends mail or calls out, autonomous transactions,
DDL that auto-commits on some backends. Only the grant can.
- **A writable driver must connect with a login limited to its write targets** — `SELECT` as above,
plus exactly what its modes need on exactly the tables / procedures its writable tags name:
`UPDATE` (Update), `INSERT` (Insert), `INSERT` + `UPDATE` (Upsert; SQL Server's `MERGE` also needs
`SELECT` on the target), `EXECUTE` on the named procedures (Procedure). Never `DELETE`, DDL or
ownership. The live gates provision exactly this (`Docker/*/seed-*.sql`) and prove a write to a
table outside the grant is refused by the server (`BadCommunicationError`) even with every driver
gate open.
- A source timestamp column is **not** touched by a write, so the next read keeps the row's old
timestamp unless the database maintains it (a trigger or `ON UPDATE CURRENT_TIMESTAMP`).
- A procedure runs with *its* rights (SQL Server `EXECUTE AS`, PostgreSQL `SECURITY DEFINER`, MySQL
`SQL SECURITY DEFINER`, Oracle definer's rights by default) — review what it can touch, since the
login's narrow grant does not bound it.
## Authoring in the AdminUI
@@ -219,6 +219,12 @@ becomes an `IAddressSpaceBuilder.Variable(browseName, displayName, DriverAttribu
### 3.4 `IWritable` — verdict: NOT in v1
> **Built since (2026-11-09).** #753 shipped the opt-in single-row UPDATE; lmxopcua#781 item 2 built the
> `write` block — but as **driver-built statements** (`mode` Update / Insert / Upsert / Procedure +
> `stampTimestamp`, procedure parameter names), not the authored `sql` text sketched below: every
> identifier goes through the catalog gate and every value is bound. The shipped contract is
> [`docs/drivers/Sql.md` → Writes](../drivers/Sql.md#writes-opt-in-753-781).
**v1 does not implement `IWritable`.** Rationale in §1. The P4 opt-in design: a tag may carry a
`write` block describing a **parameterized** statement (never string-concatenated):
@@ -43,6 +43,7 @@ public static class DraftValidator
ValidateVirtualTagTriggers(draft, errors);
ValidateSqlConnectionStringNotPersisted(draft, errors);
errors.AddRange(SqlTagDeployChecks.UndeclaredQueryErrors(draft));
errors.AddRange(SqlTagDeployChecks.WriteBlockErrors(draft));
ValidateMTConnectSingleDevice(draft, errors);
ValidateNodeAclScopes(draft, errors);
return errors;
@@ -51,6 +51,90 @@ public static class SqlTagDeployChecks
return errors;
}
/// <summary>Error code for a Sql tag whose <c>write</c> block the driver's parser refuses (lmxopcua#781 item 2).</summary>
public const string WriteBlockInvalidCode = "SqlWriteBlockInvalid";
/// <summary>Error code for a write mode the Sql driver instance's provider does not serve (lmxopcua#781 item 2).</summary>
public const string WriteModeUnsupportedCode = "SqlWriteModeUnsupported";
/// <summary>
/// Item 2. Always blocking: (a) a tag whose <c>write</c> block the driver's own parser refuses — the driver
/// would skip the whole tag (reads too) — explained by <see cref="SqlTagDefinitionFactory.DescribeWriteBlock"/>;
/// (b) a valid block whose mode the instance's <c>provider</c> does not serve
/// (<see cref="SqlWriteSupport.IsSupported"/> — Upsert / Procedure on Odbc), which the driver would publish
/// read-only. The provider is read from the same merged config the driver binds; an unreadable provider is
/// left to the driver's own construction to refuse.
/// </summary>
/// <param name="draft">The snapshot being validated.</param>
/// <returns>One error per offending tag.</returns>
public static IReadOnlyList<ValidationError> WriteBlockErrors(DraftSnapshot draft)
{
var errors = new List<ValidationError>();
foreach (var (driver, devices) in SqlInstances(draft))
{
SqlProvider? provider = null;
var providerRead = false;
foreach (var tag in draft.Tags)
{
if (!devices.Any(d => string.Equals(d.DeviceId, tag.DeviceId, StringComparison.Ordinal))) continue;
if (SqlTagDefinitionFactory.DescribeWriteBlock(tag.TagConfig) is { } invalid)
{
errors.Add(new(WriteBlockInvalidCode, $"Sql tag '{tag.TagId}': {invalid}", tag.TagId));
continue;
}
if (!SqlEquipmentTagParser.TryParse(tag.TagConfig, tag.Name, out var def) || def.Write is not { } write)
continue;
if (!providerRead)
{
provider = ReadProvider(driver, devices);
providerRead = true;
}
if (provider is { } p && !SqlWriteSupport.IsSupported(p, write.Mode))
{
errors.Add(new(WriteModeUnsupportedCode,
$"Sql tag '{tag.TagId}' writes in mode {write.Mode}, which Sql driver instance " +
$"'{driver.DriverInstanceId}' (provider {p}) does not serve — Odbc writes are Update or Insert " +
"only. Use the backend's native provider, or change the tag's 'write.mode'.",
tag.TagId));
}
}
}
return errors;
}
/// <summary>The merged config's <c>provider</c> (default SqlServer), or null when it is not a provider name.</summary>
private static SqlProvider? ReadProvider(DriverInstance driver, List<Device> devices)
{
try
{
var merged = DriverDeviceConfigMerger.Merge(
driver.DriverConfig, devices.Select(d => new DriverDeviceConfigMerger.DeviceRow(d.Name, d.DeviceConfig)).ToList(), []);
using var doc = JsonDocument.Parse(merged);
if (doc.RootElement.ValueKind != JsonValueKind.Object) return null;
foreach (var p in doc.RootElement.EnumerateObject())
{
if (!string.Equals(p.Name, "provider", StringComparison.OrdinalIgnoreCase)) continue;
return p.Value.ValueKind switch
{
JsonValueKind.String when Enum.TryParse<SqlProvider>(p.Value.GetString(), ignoreCase: true, out var v)
&& Enum.IsDefined(v) => v,
JsonValueKind.Number when p.Value.TryGetInt32(out var n) && Enum.IsDefined((SqlProvider)n) => (SqlProvider)n,
_ => null,
};
}
return SqlProvider.SqlServer;
}
catch (JsonException)
{
return null;
}
}
/// <summary>
/// Item 3. The node's write permission needs BOTH the Tag entity's <c>AccessLevel</c> (ReadWrite) and the
/// Sql blob's <c>"writable": true</c>; when they disagree the tag is silently read-only. A warning, never
@@ -30,8 +30,39 @@ public static class SqlEquipmentTagParser
/// <param name="def">The mapped definition when this returns <see langword="true"/>.</param>
/// <returns><see langword="true"/> when <paramref name="reference"/> is a valid Sql tag blob.</returns>
public static bool TryParse(string reference, string rawPath, out SqlTagDefinition def)
=> TryParseDetailed(reference, rawPath, out def, out _);
/// <summary>
/// <see cref="TryParse"/>, also reporting why an otherwise-valid tag's
/// <c>write</c> block refused it (lmxopcua#781). <paramref name="writeError"/> is null when the refusal — if
/// any — came from the tag's own fields, or when the tag parsed.
/// </summary>
/// <param name="reference">The authored equipment-tag TagConfig JSON.</param>
/// <param name="rawPath">The tag's RawPath.</param>
/// <param name="def">The mapped definition when this returns <see langword="true"/>.</param>
/// <param name="writeError">The write block's rejection reason, when that is what refused the tag.</param>
/// <returns><see langword="true"/> when <paramref name="reference"/> is a valid Sql tag blob.</returns>
public static bool TryParseDetailed(string reference, string rawPath, out SqlTagDefinition def, out string? writeError)
{
writeError = null;
if (!TryParseFields(reference, rawPath, out def, out var writeBlock)) return false;
if (writeBlock is not { } block) return true;
if (!SqlWriteSpecParser.TryParse(block, def, out var spec, out writeError))
{
def = null!;
return false;
}
def = def with { Write = spec };
return true;
}
/// <summary>The tag's own fields; the <c>write</c> block, when present, is handed back (cloned) unparsed.</summary>
private static bool TryParseFields(string reference, string rawPath, out SqlTagDefinition def, out JsonElement? writeBlock)
{
def = null!;
writeBlock = null;
if (string.IsNullOrWhiteSpace(reference)) return false;
if (reference.TrimStart().FirstOrDefault() != '{') return false;
try
@@ -55,6 +86,7 @@ public static class SqlEquipmentTagParser
: null;
var timestampColumn = ReadString(root, "timestampColumn");
if (root.TryGetProperty("write", out var writeEl)) writeBlock = writeEl.Clone();
// The per-tag write opt-in (design §3.4). Strict like the enums: absent ⇒ read-only, a JSON boolean ⇒
// that, anything else (a "true" string, a number) ⇒ the tag is refused rather than guessed at — a
@@ -43,6 +43,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
/// 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.
/// </param>
/// <param name="Write">
/// The tag's optional <c>write</c> block (lmxopcua#781 item 2): the statement shape a write becomes. Null ⇒ the
/// default single-row UPDATE (<see cref="SqlWriteSpec.Default"/>). Meaningful only when <paramref name="Writable"/>.
/// </param>
public sealed record SqlTagDefinition(
string Name,
SqlTagModel Model,
@@ -59,4 +63,9 @@ public sealed record SqlTagDefinition(
string? QueryName = null,
string? RowKeyColumn = null,
string? RowKeyValue = null,
bool Writable = false);
bool Writable = false,
SqlWriteSpec? Write = null)
{
/// <summary>The effective write spec — the authored block, or the default UPDATE.</summary>
public SqlWriteSpec EffectiveWrite => Write ?? SqlWriteSpec.Default;
}
@@ -51,7 +51,14 @@ public static class SqlTagDefinitionFactory
return warnings;
// The accept gate: whatever the driver serves draws nothing. See the class remarks.
if (SqlEquipmentTagParser.TryParse(reference, InspectionRawPath, out _)) return warnings;
if (SqlEquipmentTagParser.TryParseDetailed(reference, InspectionRawPath, out _, out var writeError)) return warnings;
// The tag's own fields parsed and its write block is what refused it: the parser said why.
if (writeError is not null)
{
warnings.Add(FormatWriteError(writeError));
return warnings;
}
try
{
@@ -83,6 +90,25 @@ public static class SqlTagDefinitionFactory
return warnings;
}
/// <summary>
/// Why the blob's <c>write</c> block (lmxopcua#781 item 2) refuses an otherwise-valid tag, or null when the
/// block is valid, absent, or the tag's own fields already refuse it (which <see cref="Inspect"/> reports).
/// The verdict is the runtime parser's own (<see cref="SqlEquipmentTagParser.TryParseDetailed"/>);
/// the deploy gate turns it into a blocking error. Never throws.
/// </summary>
/// <param name="reference">The equipment tag's TagConfig JSON.</param>
/// <returns>The operator-facing message, or null.</returns>
public static string? DescribeWriteBlock(string reference)
{
if (string.IsNullOrWhiteSpace(reference)) return null;
return SqlEquipmentTagParser.TryParseDetailed(reference, InspectionRawPath, out _, out var writeError) || writeError is null
? null
: FormatWriteError(writeError);
}
private static string FormatWriteError(string writeError)
=> $"Sql TagConfig 'write' block is invalid: {writeError} — the driver will skip this tag (BadNodeIdUnknown).";
/// <summary>
/// Explains why <see cref="SqlEquipmentTagParser.TryParse"/> refused an object-rooted blob, walking
/// its checks in the same order. Every branch here mirrors a <c>return false</c> in the parser — this
@@ -0,0 +1,275 @@
using System.Text.Json;
using System.Text.RegularExpressions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
/// <summary>The statement shape a Sql tag's write becomes (design §3.4, lmxopcua#781 item 2).</summary>
public enum SqlWriteMode
{
/// <summary>
/// <c>UPDATE … SET value WHERE key</c> on exactly one existing row — the default, and the only shape before
/// #781. Zero rows is <c>BadNoData</c>; the driver never inserts in this mode.
/// </summary>
Update,
/// <summary><c>INSERT</c> of one new row — the key (when the model has one), the value and, optionally, the stamp.</summary>
Insert,
/// <summary>
/// Update the keyed row, or insert it when absent, in one statement (SQL Server / Oracle <c>MERGE</c>,
/// PostgreSQL <c>INSERT … ON CONFLICT</c>, MySQL <c>INSERT … ON DUPLICATE KEY UPDATE</c>). Not on Odbc.
/// </summary>
Upsert,
/// <summary>A call of one named, catalog-validated stored procedure with the value (and optionally key / stamp) bound.</summary>
Procedure,
}
/// <summary>
/// The parsed form of a tag's optional <c>write</c> block. Every string is authored and NEVER reaches SQL text
/// unvalidated: <see cref="Procedure"/> is resolved against the live routine catalog by the catalog gate and
/// dialect-quoted, and the parameter names are restricted to a plain identifier charset by the parser
/// (<see cref="SqlWriteSpecParser"/>). Values are always bound.
/// </summary>
/// <param name="Mode">The statement shape.</param>
/// <param name="StampTimestamp">
/// Update / Insert / Upsert: also set the tag's timestamp column (<c>timestampColumn</c>, or a WideRow's
/// <c>topByTimestamp</c>) to the write time, in UTC.
/// </param>
/// <param name="Procedure">Procedure: the <c>procedure</c> or <c>schema.procedure</c> to call.</param>
/// <param name="ValueParameter">Procedure: the procedure parameter the written value binds to (default <c>value</c>).</param>
/// <param name="KeyParameter">Procedure: the parameter the tag's key binds to; null ⇒ the key is not passed.</param>
/// <param name="TimestampParameter">Procedure: the parameter the write time (UTC) binds to; null ⇒ not passed.</param>
public sealed record SqlWriteSpec(
SqlWriteMode Mode,
bool StampTimestamp = false,
string? Procedure = null,
string? ValueParameter = null,
string? KeyParameter = null,
string? TimestampParameter = null)
{
/// <summary>The default when a writable tag carries no <c>write</c> block: today's single-row UPDATE.</summary>
public static SqlWriteSpec Default { get; } = new(SqlWriteMode.Update);
}
/// <summary>
/// Which <see cref="SqlWriteMode"/>s each provider serves. One table, read by the dialects (runtime), the deploy
/// gate (<c>SqlTagDeployChecks</c>) and the docs, so what deploy accepts is what the driver serves.
/// <para><b>Odbc is Update / Insert only.</b> The ANSI profile has no upsert every ODBC backend accepts
/// (<c>MERGE</c> is missing or partial on several), and positional <c>?</c> markers cannot be reused, which every
/// single-statement upsert needs; a stored-procedure call would need a routine catalog and calling convention per
/// backend. Use the backend's native provider for those modes.</para>
/// </summary>
public static class SqlWriteSupport
{
/// <summary>True when <paramref name="provider"/> serves <paramref name="mode"/>.</summary>
/// <param name="provider">The driver's provider.</param>
/// <param name="mode">The tag's write mode.</param>
/// <returns>Whether the driver will execute the mode on that provider.</returns>
public static bool IsSupported(SqlProvider provider, SqlWriteMode mode) => mode switch
{
SqlWriteMode.Update or SqlWriteMode.Insert => true,
SqlWriteMode.Upsert or SqlWriteMode.Procedure => provider != SqlProvider.Odbc,
_ => false,
};
}
/// <summary>
/// Parses and validates a tag's <c>write</c> block against the tag it belongs to. Used by
/// <see cref="SqlEquipmentTagParser.TryParse"/> (a failure refuses the whole tag) and — through
/// <see cref="SqlTagDefinitionFactory.DescribeWriteBlock"/> — by the deploy gate and the AdminUI, so all three
/// share one verdict and one explanation.
/// </summary>
public static partial class SqlWriteSpecParser
{
/// <summary>The keys a <c>write</c> block may carry. Anything else is refused — a typo in a write permission is
/// the last place to guess.</summary>
private static readonly string[] KnownKeys =
["mode", "stampTimestamp", "procedure", "valueParam", "keyParam", "timestampParam"];
private static readonly string[] ProcedureOnlyKeys = ["procedure", "valueParam", "keyParam", "timestampParam"];
/// <summary>The default procedure parameter the value binds to.</summary>
public const string DefaultValueParameter = "value";
/// <summary>
/// Parses <paramref name="write"/> for <paramref name="tag"/> (whose own fields are already parsed).
/// </summary>
/// <param name="write">The <c>write</c> property's value.</param>
/// <param name="tag">The tag definition the block belongs to.</param>
/// <param name="spec">The spec, when this returns <see langword="true"/>.</param>
/// <param name="error">An operator-facing reason, when this returns <see langword="false"/>.</param>
/// <returns>Whether the block is valid for the tag.</returns>
public static bool TryParse(JsonElement write, SqlTagDefinition tag, out SqlWriteSpec? spec, out string? error)
{
ArgumentNullException.ThrowIfNull(tag);
spec = null;
error = null;
if (write.ValueKind != JsonValueKind.Object)
return Fail("'write' must be a JSON object", out error);
foreach (var property in write.EnumerateObject())
{
if (!KnownKeys.Contains(property.Name, StringComparer.Ordinal))
{
return Fail(
$"'write' has an unknown key '{property.Name}' (valid: {string.Join(", ", KnownKeys)})", out error);
}
}
var mode = SqlWriteMode.Update;
if (write.TryGetProperty("mode", out var modeEl))
{
if (modeEl.ValueKind != JsonValueKind.String
|| TagConfigJson.ClassifyEnumText(modeEl.GetString(), out mode) != JsonEnumRead.Valid)
{
return Fail(
$"'write.mode' must be one of {string.Join(", ", Enum.GetNames<SqlWriteMode>())}", out error);
}
}
var stamp = false;
if (write.TryGetProperty("stampTimestamp", out var stampEl))
{
if (stampEl.ValueKind is not (JsonValueKind.True or JsonValueKind.False))
return Fail("'write.stampTimestamp' must be a JSON boolean", out error);
stamp = stampEl.GetBoolean();
}
if (mode != SqlWriteMode.Procedure)
{
foreach (var key in ProcedureOnlyKeys)
{
if (write.TryGetProperty(key, out _))
return Fail($"'write.{key}' applies only to mode Procedure (this block's mode is {mode})", out error);
}
if (tag.Model == SqlTagModel.Query)
{
return Fail(
$"a Query tag reads a named query, which names no table to {mode.ToString().ToLowerInvariant()}; " +
"only mode Procedure can write a Query tag", out error);
}
var hasWherePair = tag.Model == SqlTagModel.KeyValue || tag.RowSelectorColumn is not null;
if (mode is SqlWriteMode.Update or SqlWriteMode.Upsert && !hasWherePair)
{
return Fail(
$"mode {mode} needs a row key — a KeyValue tag, or a WideRow 'whereColumn'/'whereValue' pair; " +
"a 'topByTimestamp' row moves under a concurrent insert (use Insert or Procedure)", out error);
}
if (stamp && StampColumn(tag) is null)
{
return Fail(
"'write.stampTimestamp' is true but the tag has no timestamp column to stamp — author " +
"'timestampColumn' (or a WideRow 'topByTimestamp')", out error);
}
spec = new SqlWriteSpec(mode, stamp);
return true;
}
// ---- Procedure ----
if (stamp)
return Fail("'write.stampTimestamp' does not apply to mode Procedure — bind the write time with 'timestampParam'", out error);
if (!write.TryGetProperty("procedure", out var procEl) || procEl.ValueKind != JsonValueKind.String
|| string.IsNullOrWhiteSpace(procEl.GetString()))
{
return Fail("mode Procedure needs 'write.procedure' — the procedure or schema.procedure to call", out error);
}
var procedure = procEl.GetString()!.Trim();
if (procedure.Split('.').Length > 2)
{
return Fail(
"'write.procedure' may be 'procedure' or 'schema.procedure' only — a package, cross-database or " +
"linked-server name cannot be validated against this connection's catalog", out error);
}
if (!TryReadParameter(write, "valueParam", DefaultValueParameter, out var valueParam, out error)) return false;
if (!TryReadParameter(write, "keyParam", null, out var keyParam, out error)) return false;
if (!TryReadParameter(write, "timestampParam", null, out var timestampParam, out error)) return false;
if (keyParam is not null && KeyValue(tag) is null)
{
return Fail(
"'write.keyParam' is set but the tag has no key to bind — a KeyValue 'keyValue', a WideRow " +
"'whereValue' or a Query 'rowKey.value'", out error);
}
var names = new[] { valueParam, keyParam, timestampParam }.Where(n => n is not null).ToArray();
if (names.Distinct(StringComparer.OrdinalIgnoreCase).Count() != names.Length)
return Fail("'write' binds two values to the same procedure parameter", out error);
spec = new SqlWriteSpec(SqlWriteMode.Procedure, false, procedure, valueParam, keyParam, timestampParam);
return true;
}
/// <summary>The column a <c>stampTimestamp</c> write sets: <c>timestampColumn</c>, else a WideRow's <c>topByTimestamp</c>.</summary>
/// <param name="tag">The tag.</param>
/// <returns>The column, or null when the tag has none.</returns>
public static string? StampColumn(SqlTagDefinition tag)
{
ArgumentNullException.ThrowIfNull(tag);
if (tag.Model == SqlTagModel.Query) return null;
if (!string.IsNullOrWhiteSpace(tag.TimestampColumn)) return tag.TimestampColumn;
return tag.Model == SqlTagModel.WideRow && !string.IsNullOrWhiteSpace(tag.RowSelectorTopByTimestamp)
? tag.RowSelectorTopByTimestamp
: null;
}
/// <summary>The tag's key value — what a procedure's <c>keyParam</c> binds: keyValue, whereValue or rowKey.value.</summary>
/// <param name="tag">The tag.</param>
/// <returns>The key, or null when the model names none.</returns>
public static string? KeyValue(SqlTagDefinition tag)
{
ArgumentNullException.ThrowIfNull(tag);
return tag.Model switch
{
SqlTagModel.KeyValue => tag.KeyValue,
SqlTagModel.WideRow => tag.RowSelectorValue,
SqlTagModel.Query => tag.RowKeyColumn is null ? null : tag.RowKeyValue,
_ => null,
};
}
/// <summary>
/// Reads a procedure parameter name: absent ⇒ <paramref name="fallback"/>; otherwise a plain identifier
/// (letter or underscore, then letters, digits, underscores; at most 128 characters), one leading <c>@</c> or
/// <c>:</c> tolerated and stripped. The name reaches SQL text unquoted (named notation), so this charset is
/// the injection boundary for it.
/// </summary>
private static bool TryReadParameter(
JsonElement write, string key, string? fallback, out string? name, out string? error)
{
name = fallback;
error = null;
if (!write.TryGetProperty(key, out var el)) return true;
if (el.ValueKind != JsonValueKind.String) return Fail($"'write.{key}' must be a string", out error);
var text = el.GetString()!.Trim();
if (text.Length > 0 && text[0] is '@' or ':') text = text[1..];
if (!ParameterName().IsMatch(text))
{
return Fail(
$"'write.{key}' must be a plain parameter name — a letter or underscore, then letters, digits or " +
"underscores (at most 128)", out error);
}
name = text;
return true;
}
private static bool Fail(string message, out string? error)
{
error = message;
return false;
}
[GeneratedRegex("^[A-Za-z_][A-Za-z0-9_]{0,127}$", RegexOptions.CultureInvariant)]
private static partial Regex ParameterName();
}
@@ -243,4 +243,65 @@ public interface ISqlDialect
/// <param name="sqlDataType">The bare SQL type family name; matched case-insensitively.</param>
/// <returns>The mapped driver data type, or <see cref="DriverDataType.String"/> when unrecognised.</returns>
DriverDataType MapColumnType(string sqlDataType);
// ---- writes (lmxopcua#781 item 2) ----
/// <summary>
/// Whether this dialect executes <paramref name="mode"/>. Defaults to the shared
/// <see cref="SqlWriteSupport"/> table for <see cref="Provider"/> — the same table the deploy gate checks, so
/// a mode deploy accepts is a mode the driver serves.
/// </summary>
/// <param name="mode">The tag's write mode.</param>
/// <returns>Whether a write in that mode is executed.</returns>
bool SupportsWriteMode(SqlWriteMode mode) => SqlWriteSupport.IsSupported(Provider, mode);
/// <summary>
/// The single-statement upsert of one keyed row, or null when this dialect has none. Every argument is
/// already dialect-quoted (identifiers) or a parameter marker — the dialect only arranges them, so nothing
/// authored and nothing a client sent reaches the text through here. A marker may appear more than once in
/// the result, so a positional-marker dialect must return null.
/// </summary>
/// <param name="quotedTable">The quoted, catalog-canonical target table.</param>
/// <param name="quotedKeyColumn">The quoted key column.</param>
/// <param name="keyMarker">The key's parameter marker.</param>
/// <param name="assignments">The quoted columns to set and their markers — the value first, then any stamp.</param>
/// <returns>The statement, or null when the dialect cannot upsert.</returns>
string? UpsertSql(
string quotedTable, string quotedKeyColumn, string keyMarker,
IReadOnlyList<(string QuotedColumn, string Marker)> assignments) => null;
/// <summary>
/// The most rows an upsert may report touching for one key. 1 everywhere except MySQL, whose
/// <c>ON DUPLICATE KEY UPDATE</c> reports 2 for "updated an existing row".
/// </summary>
int UpsertMaxRowsAffected => 1;
/// <summary>
/// The statement that calls one stored procedure, or null when this dialect cannot. Arguments are passed by
/// name (<c>EXEC p @a = @v</c>, <c>CALL p(a => @v)</c>, <c>BEGIN p(a => :p_v); END;</c>) except where the
/// backend has no named notation (MySQL: positional, in list order). Parameter names reach the text
/// unquoted; the parser restricts them to a plain identifier charset.
/// </summary>
/// <param name="quotedProcedure">The quoted, catalog-canonical procedure name.</param>
/// <param name="arguments">Procedure parameter name → marker, in order.</param>
/// <returns>The statement, or null.</returns>
string? ProcedureCallSql(string quotedProcedure, IReadOnlyList<(string Parameter, string Marker)> arguments)
=> null;
/// <summary>
/// Catalog query listing the stored procedures in one schema (column <c>ROUTINE_NAME</c>, bound like
/// <see cref="ListTablesSql"/> under the base name <c>schema</c>), or null when this dialect has no routine
/// catalog — the catalog gate then refuses every Procedure-mode tag rather than quote an unvalidated name.
/// </summary>
string? ListRoutinesSql => null;
/// <summary>
/// Classifies a write's provider exception so a constraint the database enforced reads as an authoring /
/// data fault rather than "the database is unreachable". Must never throw. The default reads the ANSI
/// <see cref="DbException.SqlState"/> when the provider reports one.
/// </summary>
/// <param name="exception">What the write threw.</param>
/// <returns>The failure class.</returns>
SqlWriteFailure ClassifyWriteFailure(Exception exception)
=> SqlWriteFailures.FromSqlState((exception as DbException)?.SqlState);
}
@@ -177,4 +177,48 @@ public sealed class MySqlDialect : ISqlDialect
_ => DriverDataType.String,
};
}
// ---- writes (lmxopcua#781 item 2) ----
/// <inheritdoc />
// INSERT … ON DUPLICATE KEY UPDATE — fires on ANY unique key of the table, so the key column must be the table's
// primary key or a unique key (and the table should carry no OTHER unique key a write could collide with).
// MySqlConnector substitutes parameters client-side, so a marker may repeat.
public string? UpsertSql(
string quotedTable, string quotedKeyColumn, string keyMarker,
IReadOnlyList<(string QuotedColumn, string Marker)> assignments)
=> string.Concat(
"INSERT INTO ", quotedTable, " (", quotedKeyColumn, ", ", string.Join(", ", assignments.Select(a => a.QuotedColumn)),
") VALUES (", keyMarker, ", ", string.Join(", ", assignments.Select(a => a.Marker)),
") ON DUPLICATE KEY UPDATE ", string.Join(", ", assignments.Select(a => a.QuotedColumn + " = " + a.Marker)));
/// <inheritdoc />
// 1 for an inserted row, 2 for an updated one (MySQL's documented ON DUPLICATE KEY UPDATE count). MySqlConnector
// defaults to found-rows (UseAffectedRows=false), so an update to the same value still reports a row, not 0.
public int UpsertMaxRowsAffected => 2;
/// <inheritdoc />
// MySQL has no named-argument notation: arguments are POSITIONAL, in the order the driver lists them — the value,
// then the key (keyParam), then the write time (timestampParam). The parameter names are not sent.
public string? ProcedureCallSql(string quotedProcedure, IReadOnlyList<(string Parameter, string Marker)> arguments)
=> "CALL " + quotedProcedure + "(" + string.Join(", ", arguments.Select(a => a.Marker)) + ")";
/// <inheritdoc />
public string? ListRoutinesSql =>
"SELECT ROUTINE_NAME AS ROUTINE_NAME FROM information_schema.ROUTINES " +
"WHERE ROUTINE_SCHEMA = @schema AND ROUTINE_TYPE = 'PROCEDURE' ORDER BY ROUTINE_NAME";
/// <inheritdoc />
// 1062 duplicate entry; 3819 CHECK violated (reported as SQLSTATE HY000, so it needs the number); the rest by
// SQLSTATE — 23000 covers NOT NULL (1048) and foreign keys (1451/1452), class 22 overflow/truncation.
public SqlWriteFailure ClassifyWriteFailure(Exception exception)
{
if (exception is not MySqlException my) return SqlWriteFailure.Other;
return my.Number switch
{
1062 => SqlWriteFailure.UniqueViolation,
3819 => SqlWriteFailure.IntegrityViolation,
_ => SqlWriteFailures.FromSqlState(my.SqlState),
};
}
}
@@ -170,4 +170,15 @@ public sealed class OdbcDialect : ISqlDialect
_ => DriverDataType.String,
};
}
// ---- writes (lmxopcua#781 item 2) ----
// Update and Insert only (SqlWriteSupport): no Upsert (positional markers cannot repeat, and MERGE is missing or
// partial on several ODBC backends) and no Procedure (no portable routine catalog or calling convention).
/// <inheritdoc />
// The driver manager reports the backend's SQLSTATE on each OdbcError.
public SqlWriteFailure ClassifyWriteFailure(Exception exception)
=> exception is OdbcException odbc && odbc.Errors.Count > 0
? SqlWriteFailures.FromSqlState(odbc.Errors[0].SQLState)
: SqlWriteFailure.Other;
}
@@ -325,4 +325,50 @@ public sealed class OracleDialect : ISqlDialect
_ => DriverDataType.String,
};
}
// ---- writes (lmxopcua#781 item 2) ----
/// <inheritdoc />
// MERGE from a constant one-row DUAL source, matching the target's key to the bound key; BindByName (set in
// ConfigureCommand) is what lets the key and value markers appear twice. No trailing semicolon — ODP.NET refuses
// one on a SQL statement.
public string? UpsertSql(
string quotedTable, string quotedKeyColumn, string keyMarker,
IReadOnlyList<(string QuotedColumn, string Marker)> assignments)
=> string.Concat(
"MERGE INTO ", quotedTable, " tgt USING (SELECT 1 AS one FROM dual) src ON (tgt.", quotedKeyColumn, " = ",
keyMarker, ") WHEN MATCHED THEN UPDATE SET ",
string.Join(", ", assignments.Select(a => "tgt." + a.QuotedColumn + " = " + a.Marker)),
" WHEN NOT MATCHED THEN INSERT (", quotedKeyColumn, ", ", string.Join(", ", assignments.Select(a => a.QuotedColumn)),
") VALUES (", keyMarker, ", ", string.Join(", ", assignments.Select(a => a.Marker)), ")");
/// <inheritdoc />
// An anonymous PL/SQL block with named notation. Unquoted parameter names fold to upper case, as Oracle's own do.
// A standalone procedure only — a package member (pkg.proc) is a three-part name the gate cannot validate.
public string? ProcedureCallSql(string quotedProcedure, IReadOnlyList<(string Parameter, string Marker)> arguments)
=> "BEGIN " + quotedProcedure
+ (arguments.Count == 0 ? string.Empty : "(" + string.Join(", ", arguments.Select(a => a.Parameter + " => " + a.Marker)) + ")")
+ "; END;";
/// <inheritdoc />
// ALL_PROCEDURES lists a standalone procedure as OBJECT_TYPE 'PROCEDURE' with OBJECT_NAME its name; it shows only
// what the principal may execute or owns.
public string? ListRoutinesSql =>
"SELECT object_name AS ROUTINE_NAME FROM ALL_PROCEDURES " +
"WHERE owner = " + ParameterMarker("schema") + " AND object_type = 'PROCEDURE' ORDER BY object_name";
/// <inheritdoc />
// ODP.NET reports no SQLSTATE; the ORA- number decides. 1 unique; 1400/1407 NULL, 2290 CHECK, 2291/2292 FK;
// 1438 precision, 12899 too large, 1722 invalid number, 18xx date conversion.
public SqlWriteFailure ClassifyWriteFailure(Exception exception)
{
if (exception is not OracleException ora) return SqlWriteFailure.Other;
return ora.Number switch
{
1 => SqlWriteFailure.UniqueViolation,
1400 or 1407 or 2290 or 2291 or 2292 => SqlWriteFailure.IntegrityViolation,
1438 or 12899 or 1722 or 1830 or 1840 or 1841 or 1843 or 1847 or 1858 or 1861 => SqlWriteFailure.DataException,
_ => SqlWriteFailure.Other,
};
}
}
@@ -267,4 +267,35 @@ public sealed class PostgresDialect : ISqlDialect
_ => DriverDataType.String,
};
}
// ---- writes (lmxopcua#781 item 2) ----
/// <inheritdoc />
// INSERT … ON CONFLICT (key) DO UPDATE — atomic under concurrency, but it REQUIRES a unique constraint or unique
// index on exactly the key column (the server refuses the statement otherwise: SQLSTATE 42P10, which surfaces as
// BadCommunicationError). EXCLUDED is the row the insert proposed, so each value marker appears once in VALUES.
// Because a conflict target is unique, the statement touches exactly one row.
public string? UpsertSql(
string quotedTable, string quotedKeyColumn, string keyMarker,
IReadOnlyList<(string QuotedColumn, string Marker)> assignments)
=> string.Concat(
"INSERT INTO ", quotedTable, " (", quotedKeyColumn, ", ", string.Join(", ", assignments.Select(a => a.QuotedColumn)),
") VALUES (", keyMarker, ", ", string.Join(", ", assignments.Select(a => a.Marker)),
") ON CONFLICT (", quotedKeyColumn, ") DO UPDATE SET ",
string.Join(", ", assignments.Select(a => a.QuotedColumn + " = EXCLUDED." + a.QuotedColumn)));
/// <inheritdoc />
// CALL with named notation (PostgreSQL 11+ procedures). Argument types must resolve by IMPLICIT cast: a Float64
// binds as double precision, so declare a numeric parameter as double precision (or cast inside the procedure),
// and a write time binds as timestamptz.
public string? ProcedureCallSql(string quotedProcedure, IReadOnlyList<(string Parameter, string Marker)> arguments)
=> "CALL " + quotedProcedure + "(" + string.Join(", ", arguments.Select(a => a.Parameter + " => " + a.Marker)) + ")";
/// <inheritdoc />
// information_schema.routines shows only routines the current role may execute or owns — the gate's question.
public string? ListRoutinesSql =>
"SELECT routine_name AS \"ROUTINE_NAME\" FROM information_schema.routines " +
"WHERE routine_schema = @schema AND routine_type = 'PROCEDURE' ORDER BY \"ROUTINE_NAME\"";
// ClassifyWriteFailure: the interface default reads PostgresException.SqlState, which Npgsql always reports.
}
@@ -222,4 +222,52 @@ public sealed class SqlServerDialect : ISqlDialect
_ => DriverDataType.String,
};
}
// ---- writes (lmxopcua#781 item 2) ----
/// <inheritdoc />
// MERGE … WITH (HOLDLOCK): the key-range lock HOLDLOCK takes is what makes the matched/not-matched decision and
// the insert one atomic step under concurrency — a bare MERGE (or UPDATE-then-INSERT) can race two writers into a
// duplicate insert. The source is a constant one-row derived table and the ON clause compares the target's key to
// the bound key directly, so a key matching several rows updates them all and the driver's row-count rule rolls it
// back. The trailing semicolon is mandatory for MERGE.
public string? UpsertSql(
string quotedTable, string quotedKeyColumn, string keyMarker,
IReadOnlyList<(string QuotedColumn, string Marker)> assignments)
=> string.Concat(
"MERGE INTO ", quotedTable, " WITH (HOLDLOCK) AS tgt USING (SELECT 1 AS one) AS src ON tgt.",
quotedKeyColumn, " = ", keyMarker,
" WHEN MATCHED THEN UPDATE SET ", string.Join(", ", assignments.Select(a => a.QuotedColumn + " = " + a.Marker)),
" WHEN NOT MATCHED THEN INSERT (", quotedKeyColumn, ", ", string.Join(", ", assignments.Select(a => a.QuotedColumn)),
") VALUES (", keyMarker, ", ", string.Join(", ", assignments.Select(a => a.Marker)), ");");
/// <inheritdoc />
// EXEC with named arguments: the left-hand @name is the procedure's own parameter, the right-hand marker the bind.
public string? ProcedureCallSql(string quotedProcedure, IReadOnlyList<(string Parameter, string Marker)> arguments)
=> arguments.Count == 0
? "EXEC " + quotedProcedure
: "EXEC " + quotedProcedure + " " + string.Join(", ", arguments.Select(a => "@" + a.Parameter + " = " + a.Marker));
/// <inheritdoc />
// INFORMATION_SCHEMA.ROUTINES lists only routines the principal may see (it holds a permission on), which is the
// question the gate asks. Procedures only — a function is not callable with EXEC semantics here.
public string? ListRoutinesSql =>
"SELECT ROUTINE_NAME FROM INFORMATION_SCHEMA.ROUTINES " +
"WHERE ROUTINE_SCHEMA = @schema AND ROUTINE_TYPE = 'PROCEDURE' ORDER BY ROUTINE_NAME";
/// <inheritdoc />
// SqlClient reports no SQLSTATE, so the engine error number decides. 2627/2601 duplicate key (constraint /
// unique index); 547 FK or CHECK, 515 NULL into NOT NULL; 8115/220/232 arithmetic overflow, 8152/2628 truncation,
// 245/8114 conversion, 241/242 date conversion or out of range.
public SqlWriteFailure ClassifyWriteFailure(Exception exception)
{
if (exception is not SqlException sql) return SqlWriteFailure.Other;
return sql.Number switch
{
2627 or 2601 => SqlWriteFailure.UniqueViolation,
547 or 515 => SqlWriteFailure.IntegrityViolation,
8115 or 220 or 232 or 8152 or 2628 or 245 or 8114 or 241 or 242 => SqlWriteFailure.DataException,
_ => SqlWriteFailure.Other,
};
}
}
@@ -0,0 +1,33 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
/// <summary>Why a write's statement failed, as far as the provider's error says (lmxopcua#781 item 2).</summary>
public enum SqlWriteFailure
{
/// <summary>Anything else — permission denied, a missing object, a broken connection. <c>BadCommunicationError</c>.</summary>
Other,
/// <summary>A unique / primary-key violation (SQLSTATE 23505, SQL Server 2627/2601, MySQL 1062, ORA-00001). <c>BadEntryExists</c>.</summary>
UniqueViolation,
/// <summary>Another integrity constraint — NOT NULL, foreign key, CHECK (SQLSTATE class 23). <c>BadOutOfRange</c>.</summary>
IntegrityViolation,
/// <summary>A data exception — overflow, truncation, an unconvertible value (SQLSTATE class 22). <c>BadOutOfRange</c>.</summary>
DataException,
}
/// <summary>SQLSTATE-based classification shared by the dialects.</summary>
public static class SqlWriteFailures
{
/// <summary>Classifies an ANSI SQLSTATE: 23505 unique, class 23 integrity, class 22 data; anything else (or null) other.</summary>
/// <param name="sqlState">The five-character SQLSTATE, or null.</param>
/// <returns>The failure class.</returns>
public static SqlWriteFailure FromSqlState(string? sqlState)
{
if (string.IsNullOrEmpty(sqlState) || sqlState.Length < 2) return SqlWriteFailure.Other;
if (sqlState == "23505") return SqlWriteFailure.UniqueViolation;
if (sqlState.StartsWith("23", StringComparison.Ordinal)) return SqlWriteFailure.IntegrityViolation;
if (sqlState.StartsWith("22", StringComparison.Ordinal)) return SqlWriteFailure.DataException;
return SqlWriteFailure.Other;
}
}
@@ -23,17 +23,23 @@ public sealed class SqlCatalog
/// <summary>Canonical <c>schema.table</c> → that relation's canonical column names.</summary>
private readonly IReadOnlyDictionary<string, IReadOnlyList<string>> _columnsByTable;
/// <summary>Stored procedures per canonical schema — loaded only for schemas a Procedure-mode tag names (lmxopcua#781).</summary>
private readonly IReadOnlyDictionary<string, IReadOnlyList<string>> _routinesBySchema;
/// <summary>Constructs a catalog snapshot.</summary>
/// <param name="defaultSchema">The schema an unqualified object name resolves to for this connection.</param>
/// <param name="schemas">Every schema the connection can see.</param>
/// <param name="tablesBySchema">Canonical schema → its tables/views.</param>
/// <param name="columnsByTable">Canonical <c>schema.table</c> → its columns.</param>
/// <param name="routinesBySchema">Canonical schema → its stored procedures; null ⇒ none loaded.</param>
internal SqlCatalog(
string defaultSchema,
IReadOnlyList<string> schemas,
IReadOnlyDictionary<string, IReadOnlyList<string>> tablesBySchema,
IReadOnlyDictionary<string, IReadOnlyList<string>> columnsByTable)
IReadOnlyDictionary<string, IReadOnlyList<string>> columnsByTable,
IReadOnlyDictionary<string, IReadOnlyList<string>>? routinesBySchema = null)
{
_routinesBySchema = routinesBySchema ?? new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal);
DefaultSchema = defaultSchema;
_schemas = schemas;
_tablesBySchema = tablesBySchema;
@@ -93,6 +99,21 @@ public sealed class SqlCatalog
&& TryMatch(columns, authored, out canonical);
}
/// <summary>
/// Resolves an authored procedure name in a canonical schema to the catalog's spelling, by the same
/// exact-then-unique-case-insensitive rule as tables and columns.
/// </summary>
/// <param name="canonicalSchema">The schema, as <see cref="TryResolveSchema"/> returned it.</param>
/// <param name="authored">The authored procedure name.</param>
/// <param name="canonical">The catalog spelling, when this returns true.</param>
/// <returns>True when exactly one procedure matches.</returns>
public bool TryResolveRoutine(string canonicalSchema, string authored, out string canonical)
{
canonical = string.Empty;
return _routinesBySchema.TryGetValue(canonicalSchema, out var routines)
&& TryMatch(routines, authored, out canonical);
}
/// <summary>The dictionary key for a canonical relation.</summary>
/// <param name="schema">The canonical schema.</param>
/// <param name="table">The canonical table.</param>
@@ -81,16 +81,29 @@ public static class SqlCatalogGate
// 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.
// A Procedure-mode write block (lmxopcua#781) names a routine, which is resolved here like any other
// identifier — for a Query tag too, whose read side needs no catalog.
SqlTagDefinition? canonical;
SqlCatalogRejection? rejection;
if (definition.Model == SqlTagModel.Query)
{
accepted.Add(definition);
canonical = definition;
rejection = null;
}
else if (!TryCanonicalize(definition, catalog, dialect, out canonical, out rejection))
{
rejected.Add(rejection!);
continue;
}
if (TryCanonicalize(definition, catalog, dialect, out var canonical, out var rejection))
accepted.Add(canonical!);
else
if (canonical!.Write is { Mode: SqlWriteMode.Procedure } write
&& !TryResolveProcedure(canonical, write, catalog, dialect, out canonical, out rejection))
{
rejected.Add(rejection!);
continue;
}
accepted.Add(canonical!);
}
return new SqlCatalogGateResult(accepted, rejected);
@@ -162,6 +175,64 @@ public static class SqlCatalogGate
return true;
}
/// <summary>The rejection <see cref="SqlCatalogRejection.Field"/> for a write block's procedure.</summary>
internal const string ProcedureField = "Write.Procedure";
/// <summary>
/// Resolves a Procedure-mode write block's routine against the catalog and substitutes its spelling
/// (lmxopcua#781). Same rules as a table: at most two parts, each renderable, schema then routine by the
/// exact-then-unique-case-insensitive match.
/// </summary>
private static bool TryResolveProcedure(
SqlTagDefinition definition,
SqlWriteSpec write,
SqlCatalog catalog,
ISqlDialect dialect,
out SqlTagDefinition? canonical,
out SqlCatalogRejection? rejection)
{
canonical = null;
rejection = null;
if (dialect.ListRoutinesSql is null)
{
rejection = new SqlCatalogRejection(definition.Name, ProcedureField,
$"the {dialect.Provider} provider has no stored-procedure catalog to validate the procedure against, " +
"so a Procedure-mode write cannot be served on it");
return false;
}
var parts = (write.Procedure ?? string.Empty).Split('.');
if (parts.Length > MaxNameParts || parts.Any(p => !IsRenderableIdentifier(p, dialect)))
{
rejection = new SqlCatalogRejection(definition.Name, ProcedureField,
"the authored procedure name is not a usable one- or two-part identifier (empty, over-long, more than " +
"two parts, or containing control or Unicode format characters); the value is withheld from this " +
"message because it cannot be safely rendered");
return false;
}
var authoredSchema = parts.Length == MaxNameParts ? parts[0] : null;
var authoredRoutine = parts[^1];
if (!catalog.TryResolveSchema(authoredSchema, out var schema))
{
rejection = new SqlCatalogRejection(definition.Name, ProcedureField,
$"schema '{authoredSchema}' does not exist (or matches more than one schema under a case-sensitive collation)");
return false;
}
if (!catalog.TryResolveRoutine(schema, authoredRoutine, out var routine))
{
rejection = new SqlCatalogRejection(definition.Name, ProcedureField,
$"stored procedure '{authoredRoutine}' does not exist in schema '{schema}', or the connecting principal " +
"may not execute it (or the name matches more than one procedure under a case-sensitive collation)");
return false;
}
canonical = definition with { Write = write with { Procedure = SqlCatalog.QualifiedKey(schema, routine) } };
return true;
}
/// <summary>Splits and resolves the authored <c>table</c> to a canonical schema + relation.</summary>
private static bool TryResolveRelation(
SqlTagDefinition definition,
@@ -37,6 +37,8 @@ internal static class SqlCatalogLoader
private const string ColumnNameColumn = "COLUMN_NAME";
/// <summary>Base name of the schema parameter every dialect's catalog SQL carries.</summary>
private const string RoutineNameColumn = "ROUTINE_NAME";
private const string SchemaParameter = "schema";
/// <summary>Base name of the table parameter <see cref="ISqlDialect.ListColumnsSql"/> carries.</summary>
@@ -50,6 +52,7 @@ internal static class SqlCatalogLoader
/// <param name="authoredTables">The distinct authored <c>table</c> strings, as written in the blobs.</param>
/// <param name="commandTimeout">Per-command server-side backstop.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <param name="authoredRoutines">Procedure-mode write targets (lmxopcua#781) whose schemas' procedures to list.</param>
/// <returns>The loaded catalog.</returns>
/// <exception cref="InvalidOperationException">The connection can see no schemas at all.</exception>
internal static async Task<SqlCatalog> LoadAsync(
@@ -57,7 +60,8 @@ internal static class SqlCatalogLoader
ISqlDialect dialect,
IReadOnlyCollection<string> authoredTables,
TimeSpan commandTimeout,
CancellationToken cancellationToken)
CancellationToken cancellationToken,
IReadOnlyCollection<string>? authoredRoutines = null)
{
var timeoutSeconds = Math.Max(1, (int)Math.Ceiling(commandTimeout.TotalSeconds));
@@ -124,7 +128,35 @@ internal static class SqlCatalogLoader
timeoutSeconds, cancellationToken).ConfigureAwait(false);
}
return new SqlCatalog(defaultSchema, schemas, tablesBySchema, columnsByTable);
// Procedure-mode write targets (lmxopcua#781): one routine listing per schema they name. A dialect with no
// routine catalog loads nothing, and the gate then refuses those tags by name.
var routinesBySchema = new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal);
if (dialect.ListRoutinesSql is { } listRoutinesSql)
{
foreach (var authored in authoredRoutines ?? [])
{
if (string.IsNullOrWhiteSpace(authored)) continue;
var parts = authored.Split('.');
if (parts.Length > 2) continue;
string schema;
if (parts.Length == 2)
{
if (!SqlCatalog.TryMatch(schemas, parts[0], out schema)) continue;
}
else
{
schema = defaultSchema;
}
if (routinesBySchema.ContainsKey(schema)) continue;
routinesBySchema[schema] = await QueryColumnAsync(
connection, dialect, listRoutinesSql, RoutineNameColumn,
command => Bind(command, dialect.ParameterBindName(SchemaParameter), schema),
timeoutSeconds, cancellationToken).ConfigureAwait(false);
}
}
return new SqlCatalog(defaultSchema, schemas, tablesBySchema, columnsByTable, routinesBySchema);
}
/// <summary>
@@ -15,11 +15,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
/// connectivity, and the <see cref="PollGroupEngine"/> subscription overlay, while
/// <see cref="SqlPollReader"/> owns the read path.
/// <para><b>Writes are opt-in at two levels and default off</b> (design §3.4, lmxopcua#753). A write needs
/// the driver's <c>allowWrites</c>, the tag's own <c>"writable": true</c>, and a model that names one row
/// (KeyValue, or a WideRow where-pair) — on top of the node's Writable flag and the caller's WriteOperate
/// role, which the server checks first. It is a parameterised single-row <c>UPDATE</c> over the tag's
/// catalog-validated identifiers (<see cref="SqlWritePlanner"/>), run in a transaction that commits only when
/// exactly one row was touched. With <c>allowWrites</c> off every write is
/// the driver's <c>allowWrites</c>, the tag's own <c>"writable": true</c>, and a target its <c>write</c> mode can
/// use (lmxopcua#781: Update / Upsert one keyed row, Insert any table-backed tag, Procedure any tag) — on top of
/// the node's Writable flag and the caller's WriteOperate role, which the server checks first. It is one
/// parameterised statement over the tag's catalog-validated identifiers (<see cref="SqlWritePlanner"/>), run in a
/// transaction that commits only when a table write touched exactly one row (a procedure call commits when it
/// does not throw). With <c>allowWrites</c> off every write is
/// <see cref="SqlStatusCodes.BadNotWritable"/> and every variable discovers as
/// <see cref="SecurityClassification.ViewOnly"/>.</para>
/// <para><b>Health is classified here, because nothing below does it.</b> The reader honours
@@ -626,17 +627,28 @@ public sealed class SqlDriver
/// Writable flag or the caller's role, which the server enforces before a write reaches the driver.
/// </summary>
private bool IsWritePermitted(SqlTagDefinition tag)
=> _options.AllowWrites && tag.Writable
&& (tag.Model == SqlTagModel.KeyValue
|| (tag.Model == SqlTagModel.WideRow && tag.RowSelectorColumn is not null));
=> _options.AllowWrites && tag.Writable && HasWriteTarget(tag) && _dialect.SupportsWriteMode(tag.EffectiveWrite.Mode);
/// <summary>
/// Whether the tag's model gives its write mode something to write (lmxopcua#781): a procedure always; an
/// insert any table-backed tag; an update or upsert one keyed row — KeyValue, or a WideRow where-pair.
/// </summary>
private static bool HasWriteTarget(SqlTagDefinition tag) => tag.EffectiveWrite.Mode switch
{
SqlWriteMode.Procedure => true,
SqlWriteMode.Insert => tag.Model is SqlTagModel.KeyValue or SqlTagModel.WideRow,
_ => tag.Model == SqlTagModel.KeyValue || (tag.Model == SqlTagModel.WideRow && tag.RowSelectorColumn is not null),
};
/// <inheritdoc />
// Each write is a parameterised single-row UPDATE (SqlWritePlanner) over the tag's CATALOG-VALIDATED
// definition — a tag the gate rejected misses here and answers BadNodeIdUnknown, so no unvalidated identifier
// can reach an UPDATE. It runs in a transaction that commits only when exactly one row was touched: zero rows
// is BadNoData (nothing to update — the driver never inserts), more than one is BadConfigurationError (the
// selector is ambiguous, the read-side rule), and both roll back. Bounded by operationTimeout like a poll. A
// write never changes driver health: its outcome is the write's answer, not a statement about the database.
// Each write is one parameterised statement in the tag's write mode (SqlWritePlanner; Update by default) over the
// tag's CATALOG-VALIDATED definition — a tag the gate rejected misses here and answers BadNodeIdUnknown, so no
// unvalidated identifier (table, column or procedure) can reach a statement. It runs in a transaction: a table
// write commits only when it touched one row (two for a MySQL upsert that updated) — zero rows is BadNoData, more
// is BadConfigurationError (the selector is ambiguous, the read-side rule), and both roll back; a procedure call
// commits when it does not throw. A constraint the database enforced maps to BadEntryExists / BadOutOfRange
// (MapWriteFailure). Bounded by operationTimeout like a poll. Never retried (WriteIdempotent=false). A write
// never changes driver health: its outcome is the write's answer, not a statement about the database.
public async Task<IReadOnlyList<WriteResult>> WriteAsync(
IReadOnlyList<WriteRequest> writes, CancellationToken cancellationToken)
{
@@ -656,7 +668,9 @@ public sealed class SqlDriver
if (!_options.AllowWrites) return SqlStatusCodes.BadNotWritable;
if (write.FullReference is null || Lookup(write.FullReference) is not { } tag)
return SqlStatusCodes.BadNodeIdUnknown;
if (!IsWritePermitted(tag)) return SqlStatusCodes.BadNotWritable;
if (!tag.Writable || !HasWriteTarget(tag)) return SqlStatusCodes.BadNotWritable;
// Deploy refuses a mode the provider cannot serve (SqlWriteModeUnsupported); this is the runtime backstop.
if (!_dialect.SupportsWriteMode(tag.EffectiveWrite.Mode)) return SqlStatusCodes.BadNotSupported;
var planned = SqlWritePlanner.TryPlan(tag, write.Value, _dialect, out var plan);
if (planned != SqlStatusCodes.Good) return planned;
@@ -683,14 +697,29 @@ public sealed class SqlDriver
catch (Exception ex)
{
// Open or execute failed. Type only on any surface; the exception (never the connection string) to the
// log sink. Nothing was committed.
_logger.LogWarning(ex, "Sql write to '{Tag}' failed against {Endpoint}: {Type}. Driver={DriverInstanceId}",
tag.Name, Endpoint, ex.GetType().Name, _driverInstanceId);
return SqlStatusCodes.BadCommunicationError;
// log sink. Nothing was committed. A constraint the database enforced is the value's fault, not the
// link's (lmxopcua#781): a duplicate key is BadEntryExists, another integrity or data rule BadOutOfRange.
// Everything else — a missing grant included — stays BadCommunicationError.
var failure = _dialect.ClassifyWriteFailure(ex);
_logger.LogWarning(ex, "Sql write to '{Tag}' failed against {Endpoint}: {Type} ({Failure}). Driver={DriverInstanceId}",
tag.Name, Endpoint, ex.GetType().Name, failure, _driverInstanceId);
return MapWriteFailure(failure);
}
}
/// <summary>Opens a connection and runs one planned UPDATE in a transaction that commits only on one row.</summary>
/// <summary>The status a classified write failure publishes (lmxopcua#781; documented in docs/drivers/Sql.md).</summary>
internal static uint MapWriteFailure(SqlWriteFailure failure) => failure switch
{
SqlWriteFailure.UniqueViolation => SqlStatusCodes.BadEntryExists,
SqlWriteFailure.IntegrityViolation or SqlWriteFailure.DataException => SqlStatusCodes.BadOutOfRange,
_ => SqlStatusCodes.BadCommunicationError,
};
/// <summary>
/// Opens a connection and runs one planned write in a transaction. A table write commits only when it touched
/// between one and <see cref="SqlWritePlan.MaxRowsAffected"/> rows; a procedure call commits whenever it did
/// not throw (its row count is the procedure's own business).
/// </summary>
private async Task<uint> ExecuteWriteAsync(SqlTagDefinition tag, SqlWritePlan plan, CancellationToken cancellationToken)
{
var connection = _factory.CreateConnection()
@@ -725,7 +754,7 @@ public sealed class SqlDriver
rows = await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
}
if (rows == 1)
if (plan.Mode == SqlWriteMode.Procedure || (rows >= 1 && rows <= plan.MaxRowsAffected))
{
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
return SqlStatusCodes.Good;
@@ -735,9 +764,9 @@ public sealed class SqlDriver
if (rows == 0) return SqlStatusCodes.BadNoData;
_logger.LogWarning(
"Sql write to '{Tag}' matched {Rows} row(s) on '{Table}' and was rolled back: a write must update " +
"Sql {Mode} write to '{Tag}' touched {Rows} row(s) on '{Table}' and was rolled back: a write must touch " +
"exactly one row. Driver={DriverInstanceId}",
tag.Name, rows, tag.Table, _driverInstanceId);
plan.Mode, tag.Name, rows, tag.Table, _driverInstanceId);
return rows > 1 ? SqlStatusCodes.BadConfigurationError : SqlStatusCodes.BadNotSupported;
}
}
@@ -1115,6 +1144,15 @@ public sealed class SqlDriver
continue;
}
if (definition.Write is { } write && !_dialect.SupportsWriteMode(write.Mode))
{
// Deploy refuses this (SqlWriteModeUnsupported); an older artifact still reads, read-only.
_logger.LogWarning(
"Sql tag {RawPath} asks for {Mode} writes, which the {Provider} provider does not serve; it " +
"is published read-only. Driver={DriverInstanceId}",
entry.RawPath, write.Mode, _dialect.Provider, _driverInstanceId);
}
table[entry.RawPath] = definition;
}
else
@@ -1197,10 +1235,17 @@ public sealed class SqlDriver
.Distinct(StringComparer.OrdinalIgnoreCase)
.ToArray();
var catalog = tables.Length == 0
// Procedure-mode write targets (lmxopcua#781) are identifiers too — a Query tag's included.
var routines = authored.Values
.Where(definition => definition.Write is { Mode: SqlWriteMode.Procedure })
.Select(definition => definition.Write!.Procedure!)
.Distinct(StringComparer.OrdinalIgnoreCase)
.ToArray();
var catalog = tables.Length == 0 && routines.Length == 0
? SqlCatalog.Empty
: await RunBoundedAsync(
token => LoadCatalogAsync(tables, token),
token => LoadCatalogAsync(tables, routines, token),
_options.OperationTimeout,
"the catalog queries did not return",
cancellationToken).ConfigureAwait(false);
@@ -1235,7 +1280,8 @@ public sealed class SqlDriver
/// <summary>Opens one connection and reads the catalog slice the authored tables need.</summary>
private async Task<SqlCatalog> LoadCatalogAsync(
IReadOnlyCollection<string> authoredTables, CancellationToken cancellationToken)
IReadOnlyCollection<string> authoredTables, IReadOnlyCollection<string> authoredRoutines,
CancellationToken cancellationToken)
{
var connection = _factory.CreateConnection()
?? throw new InvalidOperationException(
@@ -1246,7 +1292,7 @@ public sealed class SqlDriver
connection.ConnectionString = _connectionString;
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
return await SqlCatalogLoader
.LoadAsync(connection, _dialect, authoredTables, _options.CommandTimeout, cancellationToken)
.LoadAsync(connection, _dialect, authoredTables, _options.CommandTimeout, cancellationToken, authoredRoutines)
.ConfigureAwait(false);
}
}
@@ -1257,7 +1257,22 @@ public static class SqlStatusCodes
/// </summary>
public const uint BadNotWritable = 0x803B0000u;
/// <summary>The provider could not report how many rows a write touched, so it was rolled back.</summary>
/// <summary>
/// A write the database refused for a unique / primary-key violation (lmxopcua#781) — an Insert of a key that
/// exists, typically. (<c>BadEntryExists</c>, read off <c>Opc.Ua.StatusCodes</c>.)
/// </summary>
public const uint BadEntryExists = 0x809F0000u;
/// <summary>
/// A write the database refused for another integrity rule (NOT NULL, foreign key, CHECK) or a data exception
/// (overflow, truncation, an unconvertible value) — lmxopcua#781. (<c>BadOutOfRange</c>.)
/// </summary>
public const uint BadOutOfRange = 0x803C0000u;
/// <summary>
/// The provider could not report how many rows a write touched, so it was rolled back — or the dialect does not
/// serve the tag's write mode (Upsert / Procedure on Odbc, lmxopcua#781).
/// </summary>
public const uint BadNotSupported = 0x803D0000u;
/// <summary>The quality-class mask — the top two bits of an OPC UA status code.</summary>
@@ -4,94 +4,235 @@ using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
/// <summary>
/// One compiled single-row write (lmxopcua#753): a parameterised <c>UPDATE</c> whose identifiers are the
/// tag's catalog-validated ones and whose values are all bound.
/// One compiled write (lmxopcua#753, #781): a parameterised statement whose identifiers are the tag's
/// catalog-validated ones and whose values are all bound.
/// </summary>
/// <param name="SqlText">The statement — dialect-quoted identifiers and parameter markers only.</param>
/// <param name="ParameterBindNames">The bind names, in marker order (positional providers bind in this order).</param>
/// <param name="ParameterBindNames">
/// The distinct bind names, in first-marker order (a positional provider binds in this order, and only the modes
/// it supports — Update and Insert — are planned for it, where every marker appears exactly once).
/// </param>
/// <param name="Parameters">The bound values, aligned with <paramref name="ParameterBindNames"/>.</param>
/// <param name="Mode">The statement shape — decides how the executor reads the affected-row count.</param>
/// <param name="MaxRowsAffected">
/// The most rows the statement may report for a success: 1, or 2 for a MySQL upsert. Ignored for
/// <see cref="SqlWriteMode.Procedure"/>, whose row count is the procedure's business.
/// </param>
public sealed record SqlWritePlan(
string SqlText,
IReadOnlyList<string> ParameterBindNames,
IReadOnlyList<object?> Parameters);
IReadOnlyList<object?> Parameters,
SqlWriteMode Mode = SqlWriteMode.Update,
int MaxRowsAffected = 1);
/// <summary>
/// Turns one OPC UA write into a parameterised single-row <c>UPDATE</c> (design §3.4, lmxopcua#753) — pure, no
/// I/O. <b>The write surface is deliberately narrow:</b> no authored SQL, no INSERT/UPSERT/procedure, only
/// <c>UPDATE &lt;table&gt; SET &lt;value column&gt; = @v WHERE &lt;key column&gt; = @k</c> built from the
/// tag's own identifiers.
/// Turns one OPC UA write into one parameterised statement (design §3.4, lmxopcua#753 / #781 item 2) — pure, no
/// I/O. The tag's <c>write</c> block picks the shape; there is no authored write SQL:
/// <list type="bullet">
/// <item><see cref="SqlTagModel.KeyValue"/>: <c>SET valueColumn = @v WHERE keyColumn = @k</c>.</item>
/// <item><see cref="SqlTagModel.WideRow"/> with a where-pair: <c>SET columnName = @v WHERE whereColumn = @w</c>.</item>
/// <item><see cref="SqlTagModel.WideRow"/> by <c>topByTimestamp</c>, and <see cref="SqlTagModel.Query"/>:
/// refused (<see cref="SqlStatusCodes.BadNotWritable"/>) — "the newest row" moves under a concurrent
/// insert, and a named query's projection names no table to update.</item>
/// <item><see cref="SqlWriteMode.Update"/> (the default): <c>UPDATE t SET value [, stamp] WHERE key = @k</c>
/// on a KeyValue tag or a WideRow where-pair.</item>
/// <item><see cref="SqlWriteMode.Insert"/>: <c>INSERT INTO t (key, value [, stamp]) VALUES (…)</c> — no key
/// column for a WideRow <c>topByTimestamp</c> tag (an append).</item>
/// <item><see cref="SqlWriteMode.Upsert"/>: the dialect's single-statement upsert
/// (<see cref="ISqlDialect.UpsertSql"/>).</item>
/// <item><see cref="SqlWriteMode.Procedure"/>: the dialect's call of the catalog-validated procedure
/// (<see cref="ISqlDialect.ProcedureCallSql"/>) with the value, and optionally the key and write time, bound.</item>
/// </list>
/// <para><b>Injection boundary.</b> Every identifier is one the catalog gate resolved and substituted (the
/// driver plans writes only for <em>polled</em>, i.e. gate-accepted, definitions) and reaches the text through
/// <see cref="ISqlDialect.QuoteIdentifier"/>; the written value and the key are bound parameters. Nothing an
/// OPC UA client sends reaches the command text.</para>
/// <para><b>Injection boundary.</b> Every identifier is one the catalog gate resolved and substituted (the driver
/// plans writes only for gate-accepted definitions) and reaches the text through
/// <see cref="ISqlDialect.QuoteIdentifier"/>; procedure parameter names are restricted to a plain identifier
/// charset by the tag parser; the value, the key and the stamp are bound parameters. Nothing an OPC UA client
/// sends reaches the command text.</para>
/// </summary>
public static class SqlWritePlanner
{
private const string ValueBaseName = "v";
private const string KeyBaseName = "k";
private const string TimestampBaseName = "t";
/// <summary>
/// Plans a write of <paramref name="value"/> to <paramref name="tag"/>, or explains why it is refused.
/// </summary>
/// <summary>Plans a write of <paramref name="value"/> to <paramref name="tag"/>, stamping the current UTC time.</summary>
/// <param name="tag">The catalog-validated definition.</param>
/// <param name="value">The value the client wrote.</param>
/// <param name="dialect">Quoting, markers and bind names.</param>
/// <param name="dialect">Quoting, markers, bind names and the dialect-specific statements.</param>
/// <param name="plan">The plan, when this returns <see cref="SqlStatusCodes.Good"/>.</param>
/// <returns><see cref="SqlStatusCodes.Good"/>, or the status the write publishes.</returns>
public static uint TryPlan(SqlTagDefinition tag, object? value, ISqlDialect dialect, out SqlWritePlan? plan)
=> TryPlan(tag, value, dialect, DateTime.UtcNow, out plan);
/// <summary>
/// Plans a write of <paramref name="value"/> to <paramref name="tag"/>, or explains why it is refused.
/// <see cref="SqlStatusCodes.BadNotWritable"/>: the model has no target for the mode (a Query tag outside
/// Procedure, a <c>topByTimestamp</c> row for Update/Upsert). <see cref="SqlStatusCodes.BadNotSupported"/>: the
/// dialect does not serve the mode. <see cref="SqlStatusCodes.BadTypeMismatch"/>: the value will not coerce.
/// </summary>
/// <param name="tag">The catalog-validated definition.</param>
/// <param name="value">The value the client wrote.</param>
/// <param name="dialect">Quoting, markers, bind names and the dialect-specific statements.</param>
/// <param name="utcNow">The write time bound to a stamp column or <c>timestampParam</c> (UTC).</param>
/// <param name="plan">The plan, when this returns <see cref="SqlStatusCodes.Good"/>.</param>
/// <returns><see cref="SqlStatusCodes.Good"/>, or the status the write publishes.</returns>
public static uint TryPlan(
SqlTagDefinition tag, object? value, ISqlDialect dialect, DateTime utcNow, out SqlWritePlan? plan)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(dialect);
plan = null;
string setColumn, keyColumn;
string? keyValue;
switch (tag.Model)
{
case SqlTagModel.KeyValue when !Blank(tag.ValueColumn) && !Blank(tag.KeyColumn) && tag.KeyValue is not null:
setColumn = tag.ValueColumn!;
keyColumn = tag.KeyColumn!;
keyValue = tag.KeyValue;
break;
case SqlTagModel.WideRow when !Blank(tag.ColumnName) && !Blank(tag.RowSelectorColumn)
&& tag.RowSelectorValue is not null:
setColumn = tag.ColumnName!;
keyColumn = tag.RowSelectorColumn!;
keyValue = tag.RowSelectorValue;
break;
default:
return SqlStatusCodes.BadNotWritable;
}
var write = tag.EffectiveWrite;
if (!dialect.SupportsWriteMode(write.Mode)) return SqlStatusCodes.BadNotSupported;
if (!TryConvert(value, tag.DeclaredType, out var bound)) return SqlStatusCodes.BadTypeMismatch;
var stamp = DateTime.SpecifyKind(utcNow, DateTimeKind.Utc);
string sql;
try
{
sql = string.Concat(
"UPDATE ", QuoteQualifiedName(tag.Table, dialect),
" SET ", dialect.QuoteIdentifier(setColumn), " = ", dialect.ParameterMarker(ValueBaseName),
" WHERE ", dialect.QuoteIdentifier(keyColumn), " = ", dialect.ParameterMarker(KeyBaseName));
return write.Mode == SqlWriteMode.Procedure
? PlanProcedure(tag, write, bound, stamp, dialect, out plan)
: PlanTableWrite(tag, write, bound, stamp, dialect, out plan);
}
catch (ArgumentException)
{
// Unreachable for a gate-accepted definition (its identifiers came from the catalog); refuse rather
// than build a statement around a name the dialect would not quote.
plan = null;
return SqlStatusCodes.BadConfigurationError;
}
}
// Marker order: SET's value first, then WHERE's key — a positional provider (ODBC) binds in this order.
plan = new SqlWritePlan(
sql,
[dialect.ParameterBindName(ValueBaseName), dialect.ParameterBindName(KeyBaseName)],
[bound, keyValue]);
/// <summary>Update / Insert / Upsert against the tag's own table.</summary>
private static uint PlanTableWrite(
SqlTagDefinition tag, SqlWriteSpec write, object? bound, DateTime stamp, ISqlDialect dialect, out SqlWritePlan? plan)
{
plan = null;
string valueColumn;
string? keyColumn;
string? keyValue;
switch (tag.Model)
{
case SqlTagModel.KeyValue when !Blank(tag.ValueColumn) && !Blank(tag.KeyColumn) && tag.KeyValue is not null:
valueColumn = tag.ValueColumn!;
keyColumn = tag.KeyColumn!;
keyValue = tag.KeyValue;
break;
case SqlTagModel.WideRow when !Blank(tag.ColumnName) && !Blank(tag.RowSelectorColumn)
&& tag.RowSelectorValue is not null:
valueColumn = tag.ColumnName!;
keyColumn = tag.RowSelectorColumn!;
keyValue = tag.RowSelectorValue;
break;
// Newest-row selector: an Insert appends a row (no key); Update/Upsert have no row to aim at — "the
// newest row" moves under a concurrent insert.
case SqlTagModel.WideRow when write.Mode == SqlWriteMode.Insert && !Blank(tag.ColumnName):
valueColumn = tag.ColumnName!;
keyColumn = null;
keyValue = null;
break;
default:
return SqlStatusCodes.BadNotWritable;
}
var stampColumn = write.StampTimestamp ? SqlWriteSpecParser.StampColumn(tag) : null;
if (write.StampTimestamp && stampColumn is null) return SqlStatusCodes.BadConfigurationError;
var table = QuoteQualifiedName(tag.Table, dialect);
var valueMarker = dialect.ParameterMarker(ValueBaseName);
var keyMarker = dialect.ParameterMarker(KeyBaseName);
var stampMarker = dialect.ParameterMarker(TimestampBaseName);
// The value first, then the stamp: the order of the SET list and of the insert's non-key columns.
var assignments = new List<(string QuotedColumn, string Marker)> { (dialect.QuoteIdentifier(valueColumn), valueMarker) };
if (stampColumn is not null) assignments.Add((dialect.QuoteIdentifier(stampColumn), stampMarker));
var binds = new Binds(dialect);
string sql;
var maxRows = 1;
switch (write.Mode)
{
case SqlWriteMode.Update:
// Marker order: SET's value [and stamp], then WHERE's key — a positional provider binds in this order.
sql = string.Concat(
"UPDATE ", table,
" SET ", string.Join(", ", assignments.Select(a => a.QuotedColumn + " = " + a.Marker)),
" WHERE ", dialect.QuoteIdentifier(keyColumn!), " = ", keyMarker);
binds.Add(ValueBaseName, bound);
if (stampColumn is not null) binds.Add(TimestampBaseName, stamp);
binds.Add(KeyBaseName, keyValue);
break;
case SqlWriteMode.Insert:
{
var columns = new List<string>();
var markers = new List<string>();
if (keyColumn is not null)
{
columns.Add(dialect.QuoteIdentifier(keyColumn));
markers.Add(keyMarker);
binds.Add(KeyBaseName, keyValue);
}
columns.AddRange(assignments.Select(a => a.QuotedColumn));
markers.AddRange(assignments.Select(a => a.Marker));
binds.Add(ValueBaseName, bound);
if (stampColumn is not null) binds.Add(TimestampBaseName, stamp);
sql = string.Concat(
"INSERT INTO ", table, " (", string.Join(", ", columns), ") VALUES (", string.Join(", ", markers), ")");
break;
}
case SqlWriteMode.Upsert:
{
var upsert = dialect.UpsertSql(table, dialect.QuoteIdentifier(keyColumn!), keyMarker, assignments);
if (upsert is null) return SqlStatusCodes.BadNotSupported;
sql = upsert;
// A named-marker dialect (the only kind with an upsert) binds by name, so first-appearance order is
// documentation rather than load-bearing here.
binds.Add(KeyBaseName, keyValue);
binds.Add(ValueBaseName, bound);
if (stampColumn is not null) binds.Add(TimestampBaseName, stamp);
maxRows = Math.Max(1, dialect.UpsertMaxRowsAffected);
break;
}
default:
return SqlStatusCodes.BadNotWritable;
}
plan = new SqlWritePlan(sql, binds.Names, binds.Values, write.Mode, maxRows);
return SqlStatusCodes.Good;
}
/// <summary>A call of the tag's catalog-validated procedure: value, then key, then write time — each when bound.</summary>
private static uint PlanProcedure(
SqlTagDefinition tag, SqlWriteSpec write, object? bound, DateTime stamp, ISqlDialect dialect, out SqlWritePlan? plan)
{
plan = null;
if (Blank(write.Procedure)) return SqlStatusCodes.BadConfigurationError;
var binds = new Binds(dialect);
var arguments = new List<(string Parameter, string Marker)>
{
(write.ValueParameter ?? SqlWriteSpecParser.DefaultValueParameter, dialect.ParameterMarker(ValueBaseName)),
};
binds.Add(ValueBaseName, bound);
if (write.KeyParameter is { } keyParameter)
{
var key = SqlWriteSpecParser.KeyValue(tag);
if (key is null) return SqlStatusCodes.BadConfigurationError;
arguments.Add((keyParameter, dialect.ParameterMarker(KeyBaseName)));
binds.Add(KeyBaseName, key);
}
if (write.TimestampParameter is { } timestampParameter)
{
arguments.Add((timestampParameter, dialect.ParameterMarker(TimestampBaseName)));
binds.Add(TimestampBaseName, stamp);
}
var sql = dialect.ProcedureCallSql(QuoteQualifiedName(write.Procedure!, dialect), arguments);
if (sql is null) return SqlStatusCodes.BadNotSupported;
plan = new SqlWritePlan(sql, binds.Names, binds.Values, SqlWriteMode.Procedure);
return SqlStatusCodes.Good;
}
@@ -116,9 +257,23 @@ public static class SqlWritePlanner
return false;
}
/// <summary>Quotes each part of a (catalog-canonical) <c>schema.table</c> name.</summary>
/// <summary>Quotes each part of a (catalog-canonical) <c>schema.name</c>.</summary>
private static string QuoteQualifiedName(string name, ISqlDialect dialect)
=> string.Join('.', name.Split('.').Select(dialect.QuoteIdentifier));
private static bool Blank(string? value) => string.IsNullOrWhiteSpace(value);
/// <summary>The ordered (bind name, value) list a plan carries.</summary>
private sealed class Binds(ISqlDialect dialect)
{
public List<string> Names { get; } = [];
public List<object?> Values { get; } = [];
public void Add(string baseName, object? value)
{
Names.Add(dialect.ParameterBindName(baseName));
Values.Add(value);
}
}
}
@@ -0,0 +1,67 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
using ZB.MOM.WW.OtOpcUa.Configuration.Validation;
namespace ZB.MOM.WW.OtOpcUa.Configuration.Tests;
/// <summary>
/// lmxopcua#781 item 2: a Sql tag's <c>write</c> block is checked at deploy by the driver's own parser
/// (<c>SqlWriteBlockInvalid</c>), and its mode against the instance's provider (<c>SqlWriteModeUnsupported</c>).
/// Both always block.
/// </summary>
[Trait("Category", "Unit")]
public sealed class DraftValidatorSqlWriteBlockTests
{
private static string Kv(string write) =>
"""{"model":"KeyValue","table":"t","keyColumn":"k","keyValue":"a","valueColumn":"v","writable":true,"write":""" + write + "}";
private static DraftSnapshot Draft(string driverConfig, string tagConfig, string driverType = "Sql") => new()
{
GenerationId = 1, ClusterId = "c",
DriverInstances = [new DriverInstance
{ DriverInstanceId = "di", ClusterId = "c", Name = "sql", DriverType = driverType, DriverConfig = driverConfig }],
Devices = [new Device { DeviceId = "dev", DriverInstanceId = "di", Name = "d", DeviceConfig = "{}" }],
Tags = [new Tag { TagId = "t1", DeviceId = "dev", Name = "t", DataType = "Float64",
AccessLevel = TagAccessLevel.ReadWrite, TagConfig = tagConfig }],
};
[Theory]
[InlineData("""{"mode":"Merge"}""", "'write.mode'")]
[InlineData("""{"mode":"Update","oops":1}""", "unknown key 'oops'")]
[InlineData("""{"mode":"Procedure"}""", "needs 'write.procedure'")]
[InlineData("""{"stampTimestamp":true}""", "no timestamp column")]
public void A_malformed_write_block_is_a_blocking_deploy_error_with_the_parsers_reason(string write, string reason)
=> DraftValidator.Validate(Draft("{}", Kv(write))).ShouldContain(e =>
e.Code == SqlTagDeployChecks.WriteBlockInvalidCode && e.Context == "t1" && e.Message.Contains(reason));
[Theory]
[InlineData("""{"mode":"Update"}""")]
[InlineData("""{"mode":"Insert"}""")]
[InlineData("""{"mode":"Upsert"}""")]
[InlineData("""{"mode":"Procedure","procedure":"dbo.SetSp","keyParam":"key"}""")]
public void Positive_control_valid_blocks_on_SqlServer_are_quiet(string write)
=> DraftValidator.Validate(Draft("""{"provider":"SqlServer"}""", Kv(write)))
.ShouldNotContain(e => e.Code.StartsWith("SqlWrite", StringComparison.Ordinal));
[Theory]
[InlineData("""{"mode":"Upsert"}""", true)]
[InlineData("""{"mode":"Procedure","procedure":"p"}""", true)]
[InlineData("""{"mode":"Insert"}""", false)]
[InlineData("""{"mode":"Update"}""", false)]
public void Odbc_serves_update_and_insert_only(string write, bool refused)
=> DraftValidator.Validate(Draft("""{"provider":"odbc"}""", Kv(write)))
.Any(e => e.Code == SqlTagDeployChecks.WriteModeUnsupportedCode && e.Context == "t1")
.ShouldBe(refused);
[Fact]
public void An_absent_provider_is_SqlServer_so_upsert_is_accepted()
=> DraftValidator.Validate(Draft("{}", Kv("""{"mode":"Upsert"}""")))
.ShouldNotContain(e => e.Code == SqlTagDeployChecks.WriteModeUnsupportedCode);
[Fact]
public void Tags_of_other_drivers_are_not_checked()
=> DraftValidator.Validate(Draft("{}", Kv("""{"mode":"Merge"}"""), driverType: "Modbus"))
.ShouldNotContain(e => e.Code == SqlTagDeployChecks.WriteBlockInvalidCode);
}
@@ -78,10 +78,34 @@ INSERT INTO setpoints (tag_name, sp_value) VALUES
('dup', 1.0),
('dup', 2.0);
-- Write-mode targets (lmxopcua#781, SqlWriteModesLiveTests): a primary-keyed setpoint table — the key ON DUPLICATE
-- KEY UPDATE fires on — and a DEFINER procedure that journals its arguments (MySQL CALL is positional: value, key,
-- write time, the order the driver passes them).
CREATE TABLE write_targets (
tag_name varchar(128) NOT NULL PRIMARY KEY,
sp_value double NOT NULL,
stamp_ts datetime(3) NULL
) ENGINE = InnoDB;
INSERT INTO write_targets (tag_name, sp_value, stamp_ts) VALUES ('Present', 1.0, NULL);
CREATE TABLE proc_calls (
tag_name varchar(128) NOT NULL,
sp_value double NULL,
stamp_ts datetime(3) NULL
) ENGINE = InnoDB;
DELIMITER //
CREATE PROCEDURE set_target(IN p_value double, IN p_key varchar(128), IN p_ts datetime(3))
MODIFIES SQL DATA SQL SECURITY DEFINER
BEGIN
INSERT INTO proc_calls (tag_name, sp_value, stamp_ts) VALUES (p_key, p_value, p_ts);
END//
DELIMITER ;
-- ── the read-only login ──────────────────────────────────────────────────────────────────────
DROP USER IF EXISTS 'otopcua'@'%';
CREATE USER 'otopcua'@'%' IDENTIFIED BY '__APP_PASSWORD__';
GRANT SELECT ON sqlpoll.* TO 'otopcua'@'%';
GRANT EXECUTE ON FUNCTION sqlpoll.bump_counter TO 'otopcua'@'%';
GRANT UPDATE ON sqlpoll.setpoints TO 'otopcua'@'%';
GRANT INSERT, UPDATE ON sqlpoll.write_targets TO 'otopcua'@'%';
GRANT EXECUTE ON PROCEDURE sqlpoll.set_target TO 'otopcua'@'%';
FLUSH PRIVILEGES;
@@ -29,6 +29,8 @@ docker network create "$NET" >/dev/null
docker run -d --name "$PG" --network "$NET" --label project=lmxopcua \
-e POSTGRES_PASSWORD="$PW" -e POSTGRES_DB=sqlpollfixture postgres:17-alpine >/dev/null
# a fresh postgres restarts once after initdb; probing too early can catch the shutdown (seen 2026-11-09)
sleep 8
for _ in $(seq 1 60); do docker exec "$PG" pg_isready -U postgres -d sqlpollfixture >/dev/null 2>&1 && break; sleep 1; done
docker exec -i "$PG" psql -q -v ON_ERROR_STOP=1 -v app_password="$PW" -U postgres -d sqlpollfixture \
< "$PROJECT/Docker/postgres/seed-pg.sql"
@@ -72,5 +72,41 @@ VALUES (9, NULL, 1.6, TIMESTAMP '2026-07-24 10:00:01');
INSERT INTO latest_status (station_id, oven_temp, pressure, sample_ts)
VALUES (8, 210.25, 1.4, TIMESTAMP '2026-07-24 10:00:02');
-- Write-mode targets (lmxopcua#781, SqlWriteModesLiveTests): a primary-keyed setpoint table for Update / Insert /
-- MERGE, and a procedure that journals its arguments (named notation: P_VALUE / P_KEY / P_TS).
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE write_targets PURGE';
EXCEPTION
WHEN OTHERS THEN IF SQLCODE != -942 THEN RAISE; END IF;
END;
/
BEGIN
EXECUTE IMMEDIATE 'DROP TABLE proc_calls PURGE';
EXCEPTION
WHEN OTHERS THEN IF SQLCODE != -942 THEN RAISE; END IF;
END;
/
CREATE TABLE write_targets (
tag_name VARCHAR2(128) NOT NULL PRIMARY KEY,
sp_value NUMBER NOT NULL,
stamp_ts TIMESTAMP(3)
);
INSERT INTO write_targets (tag_name, sp_value, stamp_ts) VALUES ('Present', 1, NULL);
CREATE TABLE proc_calls (
tag_name VARCHAR2(128) NOT NULL,
sp_value NUMBER,
stamp_ts TIMESTAMP(3)
);
CREATE OR REPLACE PROCEDURE set_target(p_value IN NUMBER, p_key IN VARCHAR2, p_ts IN TIMESTAMP) AS
BEGIN
INSERT INTO proc_calls (tag_name, sp_value, stamp_ts) VALUES (p_key, p_value, p_ts);
END;
/
COMMIT;
EXIT;
@@ -94,8 +94,36 @@ INSERT INTO sqlpoll.int_tag_values (tag_id, num_value, sample_ts)
VALUES (101, 77.5, TIMESTAMP '2026-07-24 10:00:00'),
(102, 12.0, TIMESTAMP '2026-07-24 10:00:01');
-- Write-mode targets (lmxopcua#781, SqlWriteModesLiveTests): a primary-keyed setpoint table — the unique key is
-- what INSERT … ON CONFLICT (tag_name) needs — and a SECURITY DEFINER procedure that journals its arguments, so the
-- login needs only EXECUTE on it. p_ts is timestamptz because a procedure argument resolves by IMPLICIT cast and the
-- driver binds the write time as timestamptz.
CREATE TABLE sqlpoll.write_targets (
tag_name varchar(128) NOT NULL PRIMARY KEY,
sp_value double precision NOT NULL,
stamp_ts timestamp(3) NULL
);
INSERT INTO sqlpoll.write_targets (tag_name, sp_value, stamp_ts) VALUES ('Present', 1.0, NULL);
CREATE TABLE sqlpoll.proc_calls (
tag_name varchar(128) NOT NULL,
sp_value double precision NULL,
stamp_ts timestamptz NULL
);
CREATE PROCEDURE sqlpoll.set_target(p_value double precision, p_key varchar, p_ts timestamptz)
LANGUAGE sql SECURITY DEFINER
AS $$ INSERT INTO sqlpoll.proc_calls (tag_name, sp_value, stamp_ts) VALUES (p_key, p_value, p_ts) $$;
RESET ROLE;
-- ── grants: USAGE + SELECT and nothing else ───────────────────────────────────────────────────
GRANT USAGE ON SCHEMA sqlpoll TO otopcua;
GRANT SELECT ON ALL TABLES IN SCHEMA sqlpoll TO otopcua;
-- …except the write-mode targets (lmxopcua#781): INSERT + UPDATE on the one setpoint table, EXECUTE on the one
-- procedure. Every other table stays SELECT-only, so a write pointed there is refused by the server's own grant.
GRANT INSERT, UPDATE ON sqlpoll.write_targets TO otopcua;
REVOKE EXECUTE ON PROCEDURE sqlpoll.set_target FROM PUBLIC;
GRANT EXECUTE ON PROCEDURE sqlpoll.set_target TO otopcua;
@@ -111,6 +111,43 @@ public sealed class OdbcReadTests
columns.Select(c => c.DisplayName).ShouldBe(["tag_name", "num_value", "sample_ts"]);
}
/// <summary>
/// lmxopcua#781: the two write modes Odbc serves, through positional <c>?</c> binding — an Update that also
/// stamps (value, stamp, key in marker order), an Insert (key, value), and its duplicate surfacing the
/// backend's SQLSTATE 23505 through the driver manager as BadEntryExists. Upsert is refused, not attempted.
/// </summary>
[Fact]
public async Task Write_overOdbc_updateWithStamp_insert_andDuplicateKey()
{
SkipUnlessLive();
var key = "O-" + Guid.NewGuid().ToString("N")[..12];
static JsonObject Target(string keyValue, JsonObject write) => new()
{
["model"] = "KeyValue", ["table"] = "sqlpoll.write_targets", ["keyColumn"] = "tag_name",
["keyValue"] = keyValue, ["valueColumn"] = "sp_value", ["timestampColumn"] = "stamp_ts",
["type"] = "Float64", ["writable"] = true, ["write"] = write,
};
await using var driver = await StartAsync(null, allowWrites: true,
Tag("W/Update", Target("Present", new JsonObject { ["mode"] = "Update", ["stampTimestamp"] = true })),
Tag("W/Insert", Target(key, new JsonObject { ["mode"] = "Insert" })),
Tag("W/Upsert", Target(key, new JsonObject { ["mode"] = "Upsert" })));
var value = 500.0 + Random.Shared.Next(1, 1000);
(await WriteAsync(driver, "W/Update", value)).ShouldBe(SqlStatusCodes.Good);
Number((await driver.ReadAsync(["W/Update"], TestContext.Current.CancellationToken))[0]).ShouldBe(value);
(await WriteAsync(driver, "W/Insert", 1.25)).ShouldBe(SqlStatusCodes.Good);
Number((await driver.ReadAsync(["W/Insert"], TestContext.Current.CancellationToken))[0]).ShouldBe(1.25);
(await WriteAsync(driver, "W/Insert", 2.25)).ShouldBe(SqlStatusCodes.BadEntryExists);
(await WriteAsync(driver, "W/Upsert", 3.25)).ShouldBe(SqlStatusCodes.BadNotSupported);
}
private static async Task<uint> WriteAsync(SqlDriver driver, string reference, object? value)
=> (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken))
.ShouldHaveSingleItem().StatusCode;
private static double Number(DataValueSnapshot snapshot) => Convert.ToDouble(snapshot.Value, CultureInfo.InvariantCulture);
private static void SkipUnlessLive()
@@ -121,7 +158,11 @@ public sealed class OdbcReadTests
}
/// <summary>Builds the driver through the production factory with <c>provider: Odbc</c>.</summary>
private static async Task<SqlDriver> StartAsync(string? queriesJson, params RawTagEntry[] tags)
private static Task<SqlDriver> StartAsync(string? queriesJson, params RawTagEntry[] tags)
=> StartAsync(queriesJson, allowWrites: false, tags);
/// <summary>Builds the driver through the production factory with <c>provider: Odbc</c>.</summary>
private static async Task<SqlDriver> StartAsync(string? queriesJson, bool allowWrites, params RawTagEntry[] tags)
{
var reference = $"OdbcIt{Guid.NewGuid():N}";
Environment.SetEnvironmentVariable(SqlConnectionStringResolver.EnvironmentVariableFor(reference), ConnectionString);
@@ -137,6 +178,7 @@ public sealed class OdbcReadTests
["rawTags"] = rawTags,
};
if (queriesJson is not null) config["queries"] = JsonNode.Parse(queriesJson);
if (allowWrites) config["allowWrites"] = true;
var json = config.ToJsonString();
var driver = SqlDriverFactoryExtensions.CreateInstance("sql-odbc-integration", json);
@@ -363,7 +363,7 @@ public sealed class SqlPollServerFixture : IAsyncLifetime
{
// Views first: a view over a dropped table blocks nothing, but dropping the table under a live
// view leaves the catalog reporting a column set that no longer resolves.
foreach (var statement in SeedStatements())
foreach (var statement in SeedStatements().Concat(WriteTargetStatements()))
await ExecuteAsync(connection, statement).ConfigureAwait(false);
}
}
@@ -378,6 +378,11 @@ public sealed class SqlPollServerFixture : IAsyncLifetime
private static IEnumerable<string> SeedStatements()
{
yield return $"DROP VIEW IF EXISTS [{Schema}].[{WideRowView}];";
// lmxopcua#781 write-mode targets (SqlWriteModesLiveTests): a primary-keyed setpoint table for
// Update / Insert / Upsert and a procedure that journals its arguments into proc_calls.
yield return $"DROP PROCEDURE IF EXISTS [{Schema}].[set_target];";
yield return $"DROP TABLE IF EXISTS [{Schema}].[write_targets];";
yield return $"DROP TABLE IF EXISTS [{Schema}].[proc_calls];";
yield return $"DROP TABLE IF EXISTS [{Schema}].[{KeyValueTable}];";
yield return $"DROP TABLE IF EXISTS [{Schema}].[{WideRowTable}];";
yield return $"DROP TABLE IF EXISTS [{Schema}].[{TypeProbeTable}];";
@@ -446,6 +451,29 @@ public sealed class SqlPollServerFixture : IAsyncLifetime
}
/// <summary>Runs one non-query batch.</summary>
private static IEnumerable<string> WriteTargetStatements()
{
yield return $"""
CREATE TABLE [{Schema}].[write_targets] (
[tag_name] nvarchar(128) NOT NULL PRIMARY KEY,
[sp_value] float NOT NULL,
[stamp_ts] datetime2(3) NULL
);
""";
yield return $"INSERT INTO [{Schema}].[write_targets] ([tag_name], [sp_value], [stamp_ts]) VALUES (N'Present', 1.0, NULL);";
yield return $"""
CREATE TABLE [{Schema}].[proc_calls] (
[tag_name] nvarchar(128) NOT NULL,
[sp_value] float NULL,
[stamp_ts] datetime2(3) NULL
);
""";
yield return $"""
CREATE PROCEDURE [{Schema}].[set_target] @p_value float, @p_key nvarchar(128), @p_ts datetime2(3) AS
INSERT INTO [{Schema}].[proc_calls] ([tag_name], [sp_value], [stamp_ts]) VALUES (@p_key, @p_value, @p_ts);
""";
}
private static async Task ExecuteAsync(SqlConnection connection, string sql)
{
var command = connection.CreateCommand();
@@ -0,0 +1,273 @@
using System.Globalization;
using System.Text.Json.Nodes;
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.IntegrationTests;
/// <summary>
/// The authored <c>write</c> block (lmxopcua#781 item 2) against real servers: Update with a stamp, Insert (and
/// the duplicate-key / NOT NULL failure mapping), the dialect's Upsert, and a stored-procedure call — one suite,
/// run per backend by the sealed subclasses below, each in its backend's existing collection and env gate.
/// <para>Every seed carries the same shape (<c>Docker/*/seed-*.sql</c>, and <see cref="SqlPollServerFixture"/>
/// for SQL Server): <c>write_targets(tag_name PK, sp_value NOT NULL, stamp_ts)</c> holding one <c>Present</c>
/// row, and <c>set_target(p_value, p_key, p_ts)</c>, which journals its arguments into <c>proc_calls</c>. Each
/// test writes a key of its own, so the suite needs no cleanup grant and tolerates re-runs.</para>
/// </summary>
public abstract class SqlWriteModesLiveSuite
{
/// <summary>Why the backend is unavailable, or null.</summary>
protected abstract string? SkipReason { get; }
/// <summary>The suite login's connection string.</summary>
protected abstract string ConnectionString { get; }
/// <summary>The backend's dialect.</summary>
protected abstract ISqlDialect Dialect { get; }
/// <summary>The schema the seed lives in, as authored (the catalog gate folds its case).</summary>
protected abstract string Schema { get; }
/// <summary>The procedure as authored — deliberately in the case the backend does NOT store, so the gate's spelling substitution is exercised.</summary>
protected abstract string AuthoredProcedure { get; }
// ---- Update ----
[Fact]
public async Task Update_withStamp_setsTheValueAndTheWriteTime_onThePresentRow()
{
SkipUnlessLive();
await using var driver = await StartAsync(Target("Sp", "Present", new JsonObject { ["stampTimestamp"] = true }));
var value = 100.0 + Random.Shared.Next(1, 1000);
var before = DateTime.UtcNow.AddMinutes(-2);
(await WriteAsync(driver, "Sp", value)).ShouldBe(SqlStatusCodes.Good);
Convert.ToDouble(await ScalarAsync("SELECT sp_value FROM {0} WHERE tag_name = 'Present'", "write_targets"),
CultureInfo.InvariantCulture).ShouldBe(value);
var stamped = Convert.ToDateTime(
await ScalarAsync("SELECT stamp_ts FROM {0} WHERE tag_name = 'Present'", "write_targets"), CultureInfo.InvariantCulture);
stamped.ShouldBeGreaterThan(before);
stamped.ShouldBeLessThan(DateTime.UtcNow.AddMinutes(2));
}
[Fact]
public async Task Update_toAnAbsentKey_isBadNoData_andInsertsNothing()
{
SkipUnlessLive();
var key = NewKey();
await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Update" }));
(await WriteAsync(driver, "Sp", 1.0)).ShouldBe(SqlStatusCodes.BadNoData);
(await CountAsync("write_targets", key)).ShouldBe(0);
}
// ---- Insert ----
[Fact]
public async Task Insert_addsTheRow_andASecondInsertOfTheKey_isBadEntryExists()
{
SkipUnlessLive();
var key = NewKey();
await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Insert" }));
(await WriteAsync(driver, "Sp", 5.25)).ShouldBe(SqlStatusCodes.Good);
(await WriteAsync(driver, "Sp", 6.25)).ShouldBe(SqlStatusCodes.BadEntryExists);
(await CountAsync("write_targets", key)).ShouldBe(1);
Convert.ToDouble(await ScalarAsync($"SELECT sp_value FROM {{0}} WHERE tag_name = '{key}'", "write_targets"),
CultureInfo.InvariantCulture).ShouldBe(5.25);
}
[Fact]
public async Task Insert_ofNull_intoANotNullColumn_isBadOutOfRange()
{
SkipUnlessLive();
var key = NewKey();
await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Insert" }));
(await WriteAsync(driver, "Sp", null)).ShouldBe(SqlStatusCodes.BadOutOfRange);
(await CountAsync("write_targets", key)).ShouldBe(0);
}
// ---- Upsert ----
[Fact]
public async Task Upsert_insertsAnAbsentKey_thenUpdatesIt_andRepeatingTheSameValueStillSucceeds()
{
SkipUnlessLive();
var key = NewKey();
await using var driver = await StartAsync(Target("Sp", key, new JsonObject { ["mode"] = "Upsert", ["stampTimestamp"] = true }));
(await WriteAsync(driver, "Sp", 1.5)).ShouldBe(SqlStatusCodes.Good);
(await WriteAsync(driver, "Sp", 2.5)).ShouldBe(SqlStatusCodes.Good);
(await WriteAsync(driver, "Sp", 2.5)).ShouldBe(SqlStatusCodes.Good); // MySQL: unchanged row, found-rows = 1
(await CountAsync("write_targets", key)).ShouldBe(1);
Convert.ToDouble(await ScalarAsync($"SELECT sp_value FROM {{0}} WHERE tag_name = '{key}'", "write_targets"),
CultureInfo.InvariantCulture).ShouldBe(2.5);
Convert.ToDouble((await driver.ReadAsync(["Sp"], TestContext.Current.CancellationToken))[0].Value,
CultureInfo.InvariantCulture).ShouldBe(2.5);
}
// ---- Procedure ----
[Fact]
public async Task Procedure_isCalled_withTheValueKeyAndWriteTime()
{
SkipUnlessLive();
var key = NewKey();
await using var driver = await StartAsync(Target("Sp", "Present", new JsonObject
{
["mode"] = "Procedure", ["procedure"] = AuthoredProcedure,
["valueParam"] = "p_value", ["keyParam"] = "p_key", ["timestampParam"] = "p_ts",
}, keyValue: key));
var before = DateTime.UtcNow.AddMinutes(-2);
(await WriteAsync(driver, "Sp", 7.75)).ShouldBe(SqlStatusCodes.Good);
(await CountAsync("proc_calls", key)).ShouldBe(1);
Convert.ToDouble(await ScalarAsync($"SELECT sp_value FROM {{0}} WHERE tag_name = '{key}'", "proc_calls"),
CultureInfo.InvariantCulture).ShouldBe(7.75);
var stamp = await ScalarAsync($"SELECT stamp_ts FROM {{0}} WHERE tag_name = '{key}'", "proc_calls");
(stamp is DateTimeOffset dto ? dto.UtcDateTime : Convert.ToDateTime(stamp, CultureInfo.InvariantCulture))
.ShouldBeGreaterThan(before);
}
[Fact]
public async Task Procedure_thatDoesNotExist_isRejectedByTheCatalogGate()
{
SkipUnlessLive();
await using var driver = await StartAsync(Target("Sp", "Present", new JsonObject
{
["mode"] = "Procedure", ["procedure"] = Schema + ".no_such_proc",
}));
(await WriteAsync(driver, "Sp", 1.0)).ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
}
// ---- helpers ----
private void SkipUnlessLive()
{
if (SkipReason is { } reason) Assert.Skip(reason);
}
private static string NewKey() => "W-" + Guid.NewGuid().ToString("N")[..12];
private RawTagEntry Target(string rawPath, string key, JsonObject write, string? keyValue = null)
{
var config = new JsonObject
{
["driver"] = "Sql",
["model"] = "KeyValue",
["table"] = Schema + ".write_targets",
["keyColumn"] = "tag_name",
["keyValue"] = keyValue ?? key,
["valueColumn"] = "sp_value",
["timestampColumn"] = "stamp_ts",
["type"] = "Float64",
["writable"] = true,
["write"] = write,
};
return new RawTagEntry(rawPath, config.ToJsonString(), WriteIdempotent: false);
}
private async Task<SqlDriver> StartAsync(params RawTagEntry[] tags)
{
var driver = new SqlDriver(
new SqlDriverOptions
{
RawTags = tags,
CommandTimeout = TimeSpan.FromSeconds(10),
OperationTimeout = TimeSpan.FromSeconds(15),
AllowWrites = true,
},
driverInstanceId: "sql-write-modes-" + Dialect.Provider,
dialect: Dialect,
connectionString: ConnectionString);
try
{
await driver.InitializeAsync("{}", TestContext.Current.CancellationToken);
}
catch
{
await driver.DisposeAsync();
throw;
}
return driver;
}
private static async Task<uint> WriteAsync(SqlDriver driver, string reference, object? value)
=> (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken))
.ShouldHaveSingleItem().StatusCode;
private async Task<long> CountAsync(string table, string key)
=> Convert.ToInt64(await ScalarAsync($"SELECT COUNT(*) FROM {{0}} WHERE tag_name = '{key}'", table),
CultureInfo.InvariantCulture);
/// <summary>Runs a scalar as the suite login; <c>{0}</c> is replaced by the schema-qualified table.</summary>
private async Task<object?> ScalarAsync(string sqlFormat, string table)
{
var connection = Dialect.Factory.CreateConnection()!;
await using (connection.ConfigureAwait(false))
{
connection.ConnectionString = ConnectionString;
await connection.OpenAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
var command = connection.CreateCommand();
await using (command.ConfigureAwait(false))
{
command.CommandText = string.Format(CultureInfo.InvariantCulture, sqlFormat, Schema + "." + table);
var value = await command.ExecuteScalarAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
return value is DBNull ? null : value;
}
}
}
}
/// <summary>SQL Server leg — the seeding fixture on the shared central server (its own database).</summary>
[Collection(SqlPollServerCollection.Name)]
public sealed class SqlServerWriteModesTests(SqlPollServerFixture fixture) : SqlWriteModesLiveSuite
{
protected override string? SkipReason => fixture.SkipReason;
protected override string ConnectionString => fixture.ConnectionString;
protected override ISqlDialect Dialect { get; } = new SqlServerDialect();
protected override string Schema => SqlPollServerFixture.Schema;
protected override string AuthoredProcedure => "SQLPOLL.SET_TARGET";
}
/// <summary>PostgreSQL leg — the granted-not-owning login; upper-case authoring folds to the stored lower case.</summary>
[Collection(PostgresCollection.Name)]
public sealed class PostgresWriteModesTests(PostgresFixture fixture) : SqlWriteModesLiveSuite
{
protected override string? SkipReason => fixture.SkipReason;
protected override string ConnectionString => fixture.ConnectionString;
protected override ISqlDialect Dialect { get; } = new PostgresDialect();
protected override string Schema => PostgresFixture.Schema;
protected override string AuthoredProcedure => "SQLPOLL.SET_TARGET";
}
/// <summary>MySQL leg — positional CALL, ON DUPLICATE KEY UPDATE's 1-or-2 row count.</summary>
[Collection(MySqlCollection.Name)]
public sealed class MySqlWriteModesTests(MySqlFixture fixture) : SqlWriteModesLiveSuite
{
protected override string? SkipReason => fixture.SkipReason;
protected override string ConnectionString => fixture.ConnectionString;
protected override ISqlDialect Dialect { get; } = new MySqlDialect();
protected override string Schema => MySqlFixture.Schema;
protected override string AuthoredProcedure => "SQLPOLL.SET_TARGET";
}
/// <summary>Oracle leg — MERGE from DUAL, an anonymous PL/SQL block; lower-case authoring folds to the stored upper case.</summary>
[Collection(OracleCollection.Name)]
public sealed class OracleWriteModesTests(OracleFixture fixture) : SqlWriteModesLiveSuite
{
protected override string? SkipReason => fixture.SkipReason;
protected override string ConnectionString => fixture.ConnectionString;
protected override ISqlDialect Dialect { get; } = new OracleDialect();
protected override string Schema => OracleFixture.Schema;
protected override string AuthoredProcedure => "otopcua.set_target";
}
@@ -0,0 +1,532 @@
using System.Globalization;
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 authored <c>write</c> block (lmxopcua#781 item 2): Update (default) / Insert / Upsert / Procedure, the
/// timestamp stamp, the parser's refusals, every dialect's statement, the routine catalog gate, constraint-failure
/// status mapping, and — over SQLite — the end-to-end Insert / Upsert / stamp paths.
/// </summary>
public sealed class SqlWriteModeTests
{
private static readonly DateTime Now = new(2026, 11, 9, 12, 0, 0, DateTimeKind.Utc);
private const string KvBase =
"\"model\":\"KeyValue\",\"table\":\"plant.sp\",\"keyColumn\":\"tag\",\"keyValue\":\"L1\",\"valueColumn\":\"val\"";
private static string Kv(string write, string extra = "") =>
"{" + KvBase + extra + ",\"writable\":true,\"write\":" + write + "}";
private static SqlTagDefinition Parse(string json)
{
SqlEquipmentTagParser.TryParse(json, "T", out var def).ShouldBeTrue();
return def;
}
// ---- parse ----
[Fact]
public void NoWriteBlock_isTheDefaultUpdate()
{
var def = Parse("{" + KvBase + ",\"writable\":true}");
def.Write.ShouldBeNull();
def.EffectiveWrite.ShouldBe(SqlWriteSpec.Default);
}
[Theory]
[InlineData("Update", SqlWriteMode.Update)]
[InlineData("insert", SqlWriteMode.Insert)]
[InlineData("UPSERT", SqlWriteMode.Upsert)]
public void Mode_parses_caseInsensitively(string mode, SqlWriteMode expected)
=> Parse(Kv($$"""{"mode":"{{mode}}"}""")).Write!.Mode.ShouldBe(expected);
[Fact]
public void ModeAbsent_inABlock_isUpdate_andStampIsRead()
{
var w = Parse(Kv("""{"stampTimestamp":true}""", ",\"timestampColumn\":\"ts\"")).Write!;
w.Mode.ShouldBe(SqlWriteMode.Update);
w.StampTimestamp.ShouldBeTrue();
}
[Fact]
public void Procedure_parsesNames_stripsAMarkerPrefix_andDefaultsTheValueParam()
{
var w = Parse(Kv("""{"mode":"Procedure","procedure":"dbo.SetSp","keyParam":"@TagName","timestampParam":":ts"}""")).Write!;
w.ShouldBe(new SqlWriteSpec(SqlWriteMode.Procedure, false, "dbo.SetSp", "value", "TagName", "ts"));
}
[Fact]
public void Procedure_isTheOneModeAQueryTagMayWrite()
{
const string query = """{"model":"Query","query":"q","column":"c","rowKey":{"column":"line","value":"L1"},"writable":true,"write":""";
Parse(query + """{"mode":"Procedure","procedure":"p","keyParam":"line"}}""").Write!.KeyParameter.ShouldBe("line");
SqlEquipmentTagParser.TryParse(query + """{"mode":"Update"}}""", "T", out _).ShouldBeFalse();
SqlEquipmentTagParser.TryParse(query + """{"mode":"Insert"}}""", "T", out _).ShouldBeFalse();
}
[Theory]
[InlineData("[]", "must be a JSON object")]
[InlineData("""{"mode":"Merge"}""", "'write.mode' must be one of")]
[InlineData("""{"mode":1}""", "'write.mode' must be one of")]
[InlineData("""{"mode":"Update","stampTimestmp":true}""", "unknown key 'stampTimestmp'")]
[InlineData("""{"stampTimestamp":"yes"}""", "must be a JSON boolean")]
[InlineData("""{"stampTimestamp":true}""", "no timestamp column")]
[InlineData("""{"mode":"Update","procedure":"p"}""", "applies only to mode Procedure")]
[InlineData("""{"mode":"Procedure"}""", "needs 'write.procedure'")]
[InlineData("""{"mode":"Procedure","procedure":" "}""", "needs 'write.procedure'")]
[InlineData("""{"mode":"Procedure","procedure":"db.dbo.p"}""", "'schema.procedure' only")]
[InlineData("""{"mode":"Procedure","procedure":"p","valueParam":"v; DROP"}""", "plain parameter name")]
[InlineData("""{"mode":"Procedure","procedure":"p","keyParam":"1abc"}""", "plain parameter name")]
[InlineData("""{"mode":"Procedure","procedure":"p","keyParam":5}""", "must be a string")]
[InlineData("""{"mode":"Procedure","procedure":"p","keyParam":"Value"}""", "same procedure parameter")]
[InlineData("""{"mode":"Procedure","procedure":"p","stampTimestamp":true}""", "does not apply to mode Procedure")]
public void MalformedBlocks_refuseTheTag_andInspectGivesTheParsersReason(string write, string reason)
{
var json = Kv(write);
SqlEquipmentTagParser.TryParse(json, "T", out _).ShouldBeFalse();
SqlEquipmentTagParser.TryParseDetailed(json, "T", out _, out var writeError).ShouldBeFalse();
writeError.ShouldNotBeNull().ShouldContain(reason);
SqlTagDefinitionFactory.DescribeWriteBlock(json).ShouldNotBeNull().ShouldContain(reason);
SqlTagDefinitionFactory.Inspect(json).ShouldHaveSingleItem().ShouldContain(reason);
}
[Fact]
public void UpdateAndUpsert_needARowKey_butInsertAppendsToANewestRowTag()
{
static string Newest(string mode) =>
$$$"""{"model":"WideRow","table":"t","columnName":"c","rowSelector":{"topByTimestamp":"ts"},"writable":true,"write":{"mode":"{{{mode}}}","stampTimestamp":true}}""";
SqlTagDefinitionFactory.DescribeWriteBlock(Newest("Update")).ShouldNotBeNull().ShouldContain("needs a row key");
SqlTagDefinitionFactory.DescribeWriteBlock(Newest("Upsert")).ShouldNotBeNull().ShouldContain("needs a row key");
SqlTagDefinitionFactory.DescribeWriteBlock(Newest("Insert")).ShouldBeNull(); // positive control
}
[Fact]
public void KeyParam_onAModelWithNoKey_isRefused()
=> SqlTagDefinitionFactory.DescribeWriteBlock(
"""{"model":"Query","query":"q","column":"c","write":{"mode":"Procedure","procedure":"p","keyParam":"k"}}""")
.ShouldNotBeNull().ShouldContain("no key to bind");
[Fact]
public void DescribeWriteBlock_isQuiet_forAValidBlock_noBlock_orABlobBrokenElsewhere()
{
SqlTagDefinitionFactory.DescribeWriteBlock(Kv("""{"mode":"Upsert"}""")).ShouldBeNull();
SqlTagDefinitionFactory.DescribeWriteBlock("{" + KvBase + "}").ShouldBeNull();
// The tag's own fields are what fail (no table) — Inspect reports that, not the write block.
SqlTagDefinitionFactory.DescribeWriteBlock("""{"model":"KeyValue","write":{"mode":"Nope"}}""").ShouldBeNull();
}
// ---- shared support table ----
[Theory]
[InlineData(SqlProvider.Odbc, SqlWriteMode.Update, true)]
[InlineData(SqlProvider.Odbc, SqlWriteMode.Insert, true)]
[InlineData(SqlProvider.Odbc, SqlWriteMode.Upsert, false)]
[InlineData(SqlProvider.Odbc, SqlWriteMode.Procedure, false)]
[InlineData(SqlProvider.SqlServer, SqlWriteMode.Upsert, true)]
[InlineData(SqlProvider.Oracle, SqlWriteMode.Procedure, true)]
public void SupportTable(SqlProvider provider, SqlWriteMode mode, bool supported)
=> SqlWriteSupport.IsSupported(provider, mode).ShouldBe(supported);
/// <summary>
/// The shared table (deploy's verdict) and each real dialect's statements (runtime's) must agree: a mode the
/// table says is served has a statement, and one it refuses has none — so deploy can never accept a mode the
/// driver then refuses, or refuse one it would serve.
/// </summary>
[Fact]
public void EveryRealDialect_buildsAStatement_exactlyForTheModesTheSharedTableServes()
{
foreach (var provider in Enum.GetValues<SqlProvider>())
{
var dialect = SqlDialects.TryCreate(provider).ShouldNotBeNull();
dialect.SupportsWriteMode(SqlWriteMode.Upsert).ShouldBe(SqlWriteSupport.IsSupported(provider, SqlWriteMode.Upsert));
(dialect.UpsertSql("t", "k", "m", [("v", "n")]) is not null)
.ShouldBe(SqlWriteSupport.IsSupported(provider, SqlWriteMode.Upsert), provider.ToString());
(dialect.ProcedureCallSql("p", [("value", "m")]) is not null && dialect.ListRoutinesSql is not null)
.ShouldBe(SqlWriteSupport.IsSupported(provider, SqlWriteMode.Procedure), provider.ToString());
}
}
// ---- planner: per mode, per dialect ----
private static SqlTagDefinition KvTag(SqlWriteSpec? write, string? ts = null) =>
new("A", SqlTagModel.KeyValue, "plant.sp", KeyColumn: "tag", KeyValue: "L1", ValueColumn: "val",
TimestampColumn: ts, Writable: true, Write: write);
private static SqlWritePlan Plan(SqlTagDefinition tag, ISqlDialect dialect, object? value = null)
{
SqlWritePlanner.TryPlan(tag, value ?? 5.5, dialect, Now, out var plan).ShouldBe(SqlStatusCodes.Good);
return plan!;
}
[Fact]
public void Update_withStamp_setsValueThenStamp_thenTheKey_inMarkerOrder()
{
var plan = Plan(KvTag(new SqlWriteSpec(SqlWriteMode.Update, StampTimestamp: true), ts: "ts"), new OdbcDialect());
plan.SqlText.ShouldBe("UPDATE \"plant\".\"sp\" SET \"val\" = ?, \"ts\" = ? WHERE \"tag\" = ?");
plan.Parameters.ShouldBe([5.5, Now, "L1"]);
plan.Mode.ShouldBe(SqlWriteMode.Update);
}
[Fact]
public void Update_withoutStamp_isByteIdenticalToTheOriginalStatement()
=> Plan(KvTag(null, ts: "ts"), new SqlServerDialect()).SqlText
.ShouldBe("UPDATE [plant].[sp] SET [val] = @v WHERE [tag] = @k");
[Fact]
public void Insert_keyValue_insertsKeyValueAndStamp_inColumnOrder()
{
var plan = Plan(KvTag(new SqlWriteSpec(SqlWriteMode.Insert, StampTimestamp: true), ts: "ts"), new OdbcDialect());
plan.SqlText.ShouldBe("INSERT INTO \"plant\".\"sp\" (\"tag\", \"val\", \"ts\") VALUES (?, ?, ?)");
plan.Parameters.ShouldBe(["L1", 5.5, Now]);
}
[Fact]
public void Insert_onANewestRowTag_appendsTheValueAndItsOrderingColumn_withNoKey()
{
var tag = new SqlTagDefinition("A", SqlTagModel.WideRow, "hist", ColumnName: "temp",
RowSelectorTopByTimestamp: "sample_ts", Writable: true, Write: new SqlWriteSpec(SqlWriteMode.Insert, true));
var plan = Plan(tag, new SqlServerDialect());
plan.SqlText.ShouldBe("INSERT INTO [hist] ([temp], [sample_ts]) VALUES (@v, @t)");
plan.ParameterBindNames.ShouldBe(["@v", "@t"]);
}
[Fact]
public void Upsert_sqlServer_isAHoldlockMerge()
{
var plan = Plan(KvTag(new SqlWriteSpec(SqlWriteMode.Upsert, true), ts: "ts"), new SqlServerDialect());
plan.SqlText.ShouldBe(
"MERGE INTO [plant].[sp] WITH (HOLDLOCK) AS tgt USING (SELECT 1 AS one) AS src ON tgt.[tag] = @k " +
"WHEN MATCHED THEN UPDATE SET [val] = @v, [ts] = @t " +
"WHEN NOT MATCHED THEN INSERT ([tag], [val], [ts]) VALUES (@k, @v, @t);");
plan.ParameterBindNames.ShouldBe(["@k", "@v", "@t"]);
plan.Parameters.ShouldBe(["L1", 5.5, Now]);
plan.MaxRowsAffected.ShouldBe(1);
}
[Fact]
public void Upsert_postgres_isInsertOnConflict()
=> Plan(KvTag(new SqlWriteSpec(SqlWriteMode.Upsert)), new PostgresDialect()).SqlText.ShouldBe(
"INSERT INTO \"plant\".\"sp\" (\"tag\", \"val\") VALUES (@k, @v) ON CONFLICT (\"tag\") DO UPDATE SET \"val\" = EXCLUDED.\"val\"");
[Fact]
public void Upsert_mySql_isOnDuplicateKeyUpdate_andAllowsTwoRows()
{
var plan = Plan(KvTag(new SqlWriteSpec(SqlWriteMode.Upsert)), new MySqlDialect());
plan.SqlText.ShouldBe("INSERT INTO `plant`.`sp` (`tag`, `val`) VALUES (@k, @v) ON DUPLICATE KEY UPDATE `val` = @v");
plan.MaxRowsAffected.ShouldBe(2);
}
[Fact]
public void Upsert_oracle_isAMergeFromDual_withNamespacedBinds()
{
var plan = Plan(KvTag(new SqlWriteSpec(SqlWriteMode.Upsert)), new OracleDialect());
plan.SqlText.ShouldBe(
"MERGE INTO \"plant\".\"sp\" tgt USING (SELECT 1 AS one FROM dual) src ON (tgt.\"tag\" = :p_k) " +
"WHEN MATCHED THEN UPDATE SET tgt.\"val\" = :p_v WHEN NOT MATCHED THEN INSERT (\"tag\", \"val\") VALUES (:p_k, :p_v)");
plan.ParameterBindNames.ShouldBe(["p_k", "p_v"]);
}
[Fact]
public void Upsert_andProcedure_onOdbc_areNotSupported()
{
SqlWritePlanner.TryPlan(KvTag(new SqlWriteSpec(SqlWriteMode.Upsert)), 1.0, new OdbcDialect(), Now, out _)
.ShouldBe(SqlStatusCodes.BadNotSupported);
SqlWritePlanner.TryPlan(KvTag(new SqlWriteSpec(SqlWriteMode.Procedure, Procedure: "p", ValueParameter: "value")),
1.0, new OdbcDialect(), Now, out _).ShouldBe(SqlStatusCodes.BadNotSupported);
}
private static SqlTagDefinition ProcTag(string? key = "TagName", string? ts = "Ts") =>
KvTag(new SqlWriteSpec(SqlWriteMode.Procedure, false, "dbo.SetSp", "Value", key, ts));
[Theory]
[InlineData(SqlProvider.SqlServer, "EXEC [dbo].[SetSp] @Value = @v, @TagName = @k, @Ts = @t")]
[InlineData(SqlProvider.Postgres, "CALL \"dbo\".\"SetSp\"(Value => @v, TagName => @k, Ts => @t)")]
[InlineData(SqlProvider.MySql, "CALL `dbo`.`SetSp`(@v, @k, @t)")]
[InlineData(SqlProvider.Oracle, "BEGIN \"dbo\".\"SetSp\"(Value => :p_v, TagName => :p_k, Ts => :p_t); END;")]
public void Procedure_callsByName_orPositionallyOnMySql_valueKeyStampInThatOrder(SqlProvider provider, string sql)
{
var plan = Plan(ProcTag(), SqlDialects.TryCreate(provider)!);
plan.SqlText.ShouldBe(sql);
plan.Parameters.ShouldBe([5.5, "L1", Now]);
plan.Mode.ShouldBe(SqlWriteMode.Procedure);
}
[Fact]
public void Procedure_bindsOnlyWhatTheBlockNames()
{
var plan = Plan(ProcTag(key: null, ts: null), new SqlServerDialect());
plan.SqlText.ShouldBe("EXEC [dbo].[SetSp] @Value = @v");
plan.Parameters.ShouldBe([5.5]);
}
[Fact]
public void AHostileValue_isBound_inEveryMode()
{
const string hostile = "x'); DROP TABLE sp; --";
foreach (var write in new[]
{
new SqlWriteSpec(SqlWriteMode.Insert), new SqlWriteSpec(SqlWriteMode.Upsert),
new SqlWriteSpec(SqlWriteMode.Procedure, false, "p", "value"),
})
{
var plan = Plan(KvTag(write) with { DeclaredType = DriverDataType.String }, new SqlServerDialect(), hostile);
plan.SqlText.ShouldNotContain("DROP");
plan.Parameters.ShouldContain(hostile);
}
}
// ---- catalog gate: the procedure is an identifier like any other ----
private static SqlCatalog Catalog(params string[] procedures) => new(
"dbo", ["dbo"],
new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal) { ["dbo"] = ["sp"] },
new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal) { ["dbo.sp"] = ["tag", "val"] },
new Dictionary<string, IReadOnlyList<string>>(StringComparer.Ordinal) { ["dbo"] = procedures });
private static SqlTagDefinition GateTag(string procedure) =>
new("A", SqlTagModel.KeyValue, "sp", KeyColumn: "tag", KeyValue: "L1", ValueColumn: "val", Writable: true,
Write: new SqlWriteSpec(SqlWriteMode.Procedure, false, procedure, "value"));
[Fact]
public void Gate_resolvesTheProcedure_toTheCatalogSpelling()
{
var result = SqlCatalogGate.Apply([GateTag("setsp")], Catalog("SetSp"), new SqlServerDialect());
result.Rejected.ShouldBeEmpty();
result.Accepted.ShouldHaveSingleItem().Write!.Procedure.ShouldBe("dbo.SetSp");
}
[Fact]
public void Gate_rejectsAnUnknownProcedure_byField()
{
var result = SqlCatalogGate.Apply([GateTag("Nope")], Catalog("SetSp"), new SqlServerDialect());
result.Accepted.ShouldBeEmpty();
var rejection = result.Rejected.ShouldHaveSingleItem();
rejection.Field.ShouldBe(SqlCatalogGate.ProcedureField);
rejection.Reason.ShouldContain("'Nope' does not exist");
}
[Fact]
public void Gate_checksAQueryTagsProcedure_too()
{
var query = new SqlTagDefinition("Q", SqlTagModel.Query, string.Empty, ColumnName: "c", QueryName: "q",
Writable: true, Write: new SqlWriteSpec(SqlWriteMode.Procedure, false, "Missing", "value"));
SqlCatalogGate.Apply([query], Catalog("SetSp"), new SqlServerDialect()).Rejected.ShouldHaveSingleItem();
SqlCatalogGate.Apply([query with { Write = query.Write! with { Procedure = "SetSp" } }], Catalog("SetSp"),
new SqlServerDialect()).Accepted.ShouldHaveSingleItem().Write!.Procedure.ShouldBe("dbo.SetSp");
}
[Fact]
public void Gate_refusesAProcedure_onADialectWithNoRoutineCatalog()
=> SqlCatalogGate.Apply([GateTag("SetSp")], Catalog("SetSp"), new OdbcDialect())
.Rejected.ShouldHaveSingleItem().Reason.ShouldContain("no stored-procedure catalog");
// ---- failure classification → status ----
[Theory]
[InlineData("23505", SqlWriteFailure.UniqueViolation)]
[InlineData("23502", SqlWriteFailure.IntegrityViolation)]
[InlineData("23000", SqlWriteFailure.IntegrityViolation)]
[InlineData("22003", SqlWriteFailure.DataException)]
[InlineData("42501", SqlWriteFailure.Other)] // insufficient privilege stays a communication error
[InlineData(null, SqlWriteFailure.Other)]
public void SqlState_classification(string? sqlState, SqlWriteFailure expected)
=> SqlWriteFailures.FromSqlState(sqlState).ShouldBe(expected);
[Theory]
[InlineData(SqlWriteFailure.UniqueViolation, 0x809F0000u)] // BadEntryExists
[InlineData(SqlWriteFailure.IntegrityViolation, 0x803C0000u)] // BadOutOfRange
[InlineData(SqlWriteFailure.DataException, 0x803C0000u)]
[InlineData(SqlWriteFailure.Other, 0x80050000u)] // BadCommunicationError
public void Failure_mapsToTheDocumentedStatus(SqlWriteFailure failure, uint status)
=> SqlDriver.MapWriteFailure(failure).ShouldBe(status);
[Fact]
public void AnUnrelatedException_isNeverAConstraint_onAnyDialect()
{
foreach (var provider in Enum.GetValues<SqlProvider>())
SqlDialects.TryCreate(provider)!.ClassifyWriteFailure(new InvalidOperationException()).ShouldBe(SqlWriteFailure.Other);
}
// ---- end to end over SQLite ----
private const string Setpoints = "Setpoints";
private static SqlitePollFixture SetpointFixture()
{
var fixture = new SqlitePollFixture();
fixture.Execute($"CREATE TABLE {Setpoints} (tag TEXT NOT NULL PRIMARY KEY, sp REAL NOT NULL, ts TEXT NULL)");
fixture.Execute($"INSERT INTO {Setpoints} VALUES ('Present', 1.0, NULL)");
return fixture;
}
private static string SpTag(string key, string write) => $$$"""
{"model":"KeyValue","table":"{{{Setpoints}}}","keyColumn":"tag","keyValue":"{{{key}}}","valueColumn":"sp",
"timestampColumn":"ts","type":"Float64","writable":true,"write":{{{write}}}}
""";
[Fact]
public async Task Insert_addsANewRow_andASecondInsertOfTheKey_isBadEntryExists()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture, Entry("New", SpTag("New", """{"mode":"Insert"}""")));
(await WriteAsync(driver, "New", 7.5)).ShouldBe(SqlStatusCodes.Good);
Scalar(fixture, "SELECT sp FROM Setpoints WHERE tag = 'New'").ShouldBe(7.5);
(await WriteAsync(driver, "New", 8.5)).ShouldBe(SqlStatusCodes.BadEntryExists);
Scalar(fixture, "SELECT sp FROM Setpoints WHERE tag = 'New'").ShouldBe(7.5);
}
[Fact]
public async Task Insert_violatingNotNull_isBadOutOfRange_andCommitsNothing()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture, Entry("New", SpTag("New", """{"mode":"Insert"}""")));
(await WriteAsync(driver, "New", null)).ShouldBe(SqlStatusCodes.BadOutOfRange);
Scalar(fixture, "SELECT COUNT(*) FROM Setpoints").ShouldBe(1.0);
}
[Fact]
public async Task Upsert_insertsAnAbsentKey_andUpdatesAPresentOne()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture,
Entry("Absent", SpTag("Absent", """{"mode":"Upsert"}""")),
Entry("Present", SpTag("Present", """{"mode":"Upsert"}""")));
(await WriteAsync(driver, "Absent", 3.0)).ShouldBe(SqlStatusCodes.Good);
(await WriteAsync(driver, "Present", 4.0)).ShouldBe(SqlStatusCodes.Good);
Scalar(fixture, "SELECT sp FROM Setpoints WHERE tag = 'Absent'").ShouldBe(3.0);
Scalar(fixture, "SELECT sp FROM Setpoints WHERE tag = 'Present'").ShouldBe(4.0);
Scalar(fixture, "SELECT COUNT(*) FROM Setpoints").ShouldBe(2.0);
}
/// <summary>The same absent key under the default Update is BadNoData and inserts nothing — the back-compat pin.</summary>
[Fact]
public async Task Update_toAnAbsentKey_isStillBadNoData()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture, Entry("Absent", SpTag("Absent", """{"mode":"Update"}""")));
(await WriteAsync(driver, "Absent", 3.0)).ShouldBe(SqlStatusCodes.BadNoData);
Scalar(fixture, "SELECT COUNT(*) FROM Setpoints").ShouldBe(1.0);
}
[Fact]
public async Task StampTimestamp_setsTheTimestampColumn_toTheWriteTime_andWithoutItTheColumnIsUntouched()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture,
Entry("Stamped", SpTag("Present", """{"mode":"Update","stampTimestamp":true}""")));
var before = DateTime.UtcNow.AddSeconds(-1);
(await WriteAsync(driver, "Stamped", 9.0)).ShouldBe(SqlStatusCodes.Good);
using var command = fixture.Connection.CreateCommand();
command.CommandText = "SELECT ts FROM Setpoints WHERE tag = 'Present'";
var stamped = DateTime.Parse((string)command.ExecuteScalar()!, CultureInfo.InvariantCulture,
DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeUniversal);
stamped.ShouldBeGreaterThan(before);
stamped.ShouldBeLessThan(DateTime.UtcNow.AddSeconds(1));
}
[Fact]
public async Task Procedure_onADialectWithNoRoutineCatalog_isRejectedByTheGate_notExecuted()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture,
Entry("Proc", SpTag("Present", """{"mode":"Procedure","procedure":"SetSp"}""")));
(await WriteAsync(driver, "Proc", 2.0)).ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
Scalar(fixture, "SELECT sp FROM Setpoints WHERE tag = 'Present'").ShouldBe(1.0);
}
[Fact]
public async Task Discover_advertisesOperate_forAnInsertOnANewestRowTag()
{
using var fixture = SetpointFixture();
await using var driver = await StartAsync(fixture, Entry("Append", $$$"""
{"model":"WideRow","table":"{{{SqlitePollFixture.WideRowTable}}}","columnName":"oven_temp",
"rowSelector":{"topByTimestamp":"sample_ts"},"writable":true,"write":{"mode":"Insert","stampTimestamp":true}}
"""));
var builder = new Capture();
await driver.DiscoverAsync(builder, CancellationToken.None);
builder.Classes["Append"].ShouldBe(SecurityClassification.Operate);
}
// ---- helpers ----
private static RawTagEntry Entry(string rawPath, string json) => new(rawPath, json, WriteIdempotent: false);
private static async Task<uint> WriteAsync(SqlDriver driver, string reference, object? value)
=> (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], CancellationToken.None))
.ShouldHaveSingleItem().StatusCode;
private static double Scalar(SqlitePollFixture fixture, string sql)
{
using var command = fixture.Connection.CreateCommand();
command.CommandText = sql;
return Convert.ToDouble(command.ExecuteScalar(), CultureInfo.InvariantCulture);
}
private static async Task<SqlDriver> StartAsync(SqlitePollFixture fixture, params RawTagEntry[] tags)
{
var driver = new SqlDriver(
new SqlDriverOptions
{
RawTags = tags,
OperationTimeout = TimeSpan.FromSeconds(15),
CommandTimeout = TimeSpan.FromSeconds(10),
AllowWrites = true,
},
"sql-write-modes", new SqliteDialect(), fixture.ConnectionString, fixture.Factory);
await driver.InitializeAsync("{}", CancellationToken.None);
return driver;
}
private sealed class Capture : IAddressSpaceBuilder
{
public Dictionary<string, SecurityClassification> Classes { get; } = [];
public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
{
Classes[attributeInfo.FullName] = attributeInfo.SecurityClass;
return new Handle(attributeInfo.FullName);
}
public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
private sealed class Handle(string fullReference) : IVariableHandle
{
public string FullReference => fullReference;
public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new Sink();
private sealed class Sink : IAlarmConditionSink
{
public void OnTransition(AlarmEventArgs args) { }
}
}
}
}
@@ -200,4 +200,28 @@ public sealed class SqliteDialect : ISqlDialect
_ => DriverDataType.String,
};
}
// ---- writes (lmxopcua#781) — SQLite stands in for a named-marker backend with a PostgreSQL-shaped upsert ----
/// <inheritdoc />
// Provider is Odbc (a stand-in), whose shared table refuses Upsert; SQLite itself upserts, so the end-to-end
// upsert suite runs offline. Procedure stays refused — SQLite has none.
public bool SupportsWriteMode(SqlWriteMode mode) => mode is SqlWriteMode.Update or SqlWriteMode.Insert or SqlWriteMode.Upsert;
/// <inheritdoc />
public string? UpsertSql(
string quotedTable, string quotedKeyColumn, string keyMarker,
IReadOnlyList<(string QuotedColumn, string Marker)> assignments)
=> string.Concat(
"INSERT INTO ", quotedTable, " (", quotedKeyColumn, ", ", string.Join(", ", assignments.Select(a => a.QuotedColumn)),
") VALUES (", keyMarker, ", ", string.Join(", ", assignments.Select(a => a.Marker)),
") ON CONFLICT (", quotedKeyColumn, ") DO UPDATE SET ",
string.Join(", ", assignments.Select(a => a.QuotedColumn + " = excluded." + a.QuotedColumn)));
/// <inheritdoc />
// SQLITE_CONSTRAINT (19): extended 2067 UNIQUE / 1555 PRIMARYKEY ⇒ unique; any other constraint ⇒ integrity.
public SqlWriteFailure ClassifyWriteFailure(Exception exception)
=> exception is SqliteException { SqliteErrorCode: 19 } sqlite
? sqlite.SqliteExtendedErrorCode is 2067 or 1555 ? SqlWriteFailure.UniqueViolation : SqlWriteFailure.IntegrityViolation
: SqlWriteFailure.Other;
}
@@ -353,4 +353,27 @@ public sealed class SqlTagConfigModelTests
def.RowSelectorColumn.ShouldBeNull();
def.RowSelectorValue.ShouldBeNull();
}
// ---- lmxopcua#781 item 2: the authored write block ------------------------------------------
[Fact]
public void A_write_block_survives_load_save_and_parses_to_the_same_spec()
{
const string blob =
"""{"model":"KeyValue","table":"t","keyColumn":"k","keyValue":"a","valueColumn":"v","timestampColumn":"ts","writable":true,"write":{"mode":"Upsert","stampTimestamp":true}}""";
var m = SqlTagConfigModel.FromJson(blob);
m.Validate().ShouldBeNull();
SqlEquipmentTagParser.TryParse(m.ToJson(), "raw/path", out var def).ShouldBeTrue();
def.Write.ShouldBe(new SqlWriteSpec(SqlWriteMode.Upsert, StampTimestamp: true));
}
[Fact]
public void A_malformed_write_block_fails_validation_with_the_runtime_parsers_reason()
{
var m = SqlTagConfigModel.FromJson(
"""{"model":"KeyValue","table":"t","keyColumn":"k","keyValue":"a","valueColumn":"v","writable":true,"write":{"mode":"Procedure"}}""");
m.Validate().ShouldNotBeNull().ShouldContain("needs 'write.procedure'");
}
}