diff --git a/docs/drivers/README.md b/docs/drivers/README.md index 124bbcc2..3dd5309f 100644 --- a/docs/drivers/README.md +++ b/docs/drivers/README.md @@ -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. | diff --git a/docs/drivers/Sql.md b/docs/drivers/Sql.md index 59ae77cf..8546d037 100644 --- a/docs/drivers/Sql.md +++ b/docs/drivers/Sql.md @@ -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 SET = @v WHERE = @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 diff --git a/docs/plans/2026-07-15-sql-poll-driver-design.md b/docs/plans/2026-07-15-sql-poll-driver-design.md index 62a13701..6d6f5e89 100644 --- a/docs/plans/2026-07-15-sql-poll-driver-design.md +++ b/docs/plans/2026-07-15-sql-poll-driver-design.md @@ -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): diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs index c01c6675..a4ac7452 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/DraftValidator.cs @@ -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; diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlTagDeployChecks.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlTagDeployChecks.cs index bdabcc4a..a0a1ad2f 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlTagDeployChecks.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Configuration/Validation/SqlTagDeployChecks.cs @@ -51,6 +51,90 @@ public static class SqlTagDeployChecks return errors; } + /// Error code for a Sql tag whose write block the driver's parser refuses (lmxopcua#781 item 2). + public const string WriteBlockInvalidCode = "SqlWriteBlockInvalid"; + + /// Error code for a write mode the Sql driver instance's provider does not serve (lmxopcua#781 item 2). + public const string WriteModeUnsupportedCode = "SqlWriteModeUnsupported"; + + /// + /// Item 2. Always blocking: (a) a tag whose write block the driver's own parser refuses — the driver + /// would skip the whole tag (reads too) — explained by ; + /// (b) a valid block whose mode the instance's provider does not serve + /// ( — 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. + /// + /// The snapshot being validated. + /// One error per offending tag. + public static IReadOnlyList WriteBlockErrors(DraftSnapshot draft) + { + var errors = new List(); + 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; + } + + /// The merged config's provider (default SqlServer), or null when it is not a provider name. + private static SqlProvider? ReadProvider(DriverInstance driver, List 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(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; + } + } + /// /// Item 3. The node's write permission needs BOTH the Tag entity's AccessLevel (ReadWrite) and the /// Sql blob's "writable": true; when they disagree the tag is silently read-only. A warning, never diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlEquipmentTagParser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlEquipmentTagParser.cs index bd3ef307..b6265d38 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlEquipmentTagParser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlEquipmentTagParser.cs @@ -30,8 +30,39 @@ public static class SqlEquipmentTagParser /// The mapped definition when this returns . /// when is a valid Sql tag blob. public static bool TryParse(string reference, string rawPath, out SqlTagDefinition def) + => TryParseDetailed(reference, rawPath, out def, out _); + + /// + /// , also reporting why an otherwise-valid tag's + /// write block refused it (lmxopcua#781). is null when the refusal — if + /// any — came from the tag's own fields, or when the tag parsed. + /// + /// The authored equipment-tag TagConfig JSON. + /// The tag's RawPath. + /// The mapped definition when this returns . + /// The write block's rejection reason, when that is what refused the tag. + /// when is a valid Sql tag blob. + 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; + } + + /// The tag's own fields; the write block, when present, is handed back (cloned) unparsed. + 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 diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinition.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinition.cs index 7144cf7f..d987c8f7 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinition.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinition.cs @@ -43,6 +43,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts; /// Optional explicit type override from the blob's type field. ⇒ the /// driver infers the type from the result set's column metadata. /// +/// +/// The tag's optional write block (lmxopcua#781 item 2): the statement shape a write becomes. Null ⇒ the +/// default single-row UPDATE (). Meaningful only when . +/// 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) +{ + /// The effective write spec — the authored block, or the default UPDATE. + public SqlWriteSpec EffectiveWrite => Write ?? SqlWriteSpec.Default; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinitionFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinitionFactory.cs index 2eb8b9c5..a5db79cf 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinitionFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlTagDefinitionFactory.cs @@ -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; } + /// + /// Why the blob's write 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 reports). + /// The verdict is the runtime parser's own (); + /// the deploy gate turns it into a blocking error. Never throws. + /// + /// The equipment tag's TagConfig JSON. + /// The operator-facing message, or null. + 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)."; + /// /// Explains why refused an object-rooted blob, walking /// its checks in the same order. Every branch here mirrors a return false in the parser — this diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlWriteSpec.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlWriteSpec.cs new file mode 100644 index 00000000..cac094a5 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts/SqlWriteSpec.cs @@ -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; + +/// The statement shape a Sql tag's write becomes (design §3.4, lmxopcua#781 item 2). +public enum SqlWriteMode +{ + /// + /// UPDATE … SET value WHERE key on exactly one existing row — the default, and the only shape before + /// #781. Zero rows is BadNoData; the driver never inserts in this mode. + /// + Update, + + /// INSERT of one new row — the key (when the model has one), the value and, optionally, the stamp. + Insert, + + /// + /// Update the keyed row, or insert it when absent, in one statement (SQL Server / Oracle MERGE, + /// PostgreSQL INSERT … ON CONFLICT, MySQL INSERT … ON DUPLICATE KEY UPDATE). Not on Odbc. + /// + Upsert, + + /// A call of one named, catalog-validated stored procedure with the value (and optionally key / stamp) bound. + Procedure, +} + +/// +/// The parsed form of a tag's optional write block. Every string is authored and NEVER reaches SQL text +/// unvalidated: 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 +/// (). Values are always bound. +/// +/// The statement shape. +/// +/// Update / Insert / Upsert: also set the tag's timestamp column (timestampColumn, or a WideRow's +/// topByTimestamp) to the write time, in UTC. +/// +/// Procedure: the procedure or schema.procedure to call. +/// Procedure: the procedure parameter the written value binds to (default value). +/// Procedure: the parameter the tag's key binds to; null ⇒ the key is not passed. +/// Procedure: the parameter the write time (UTC) binds to; null ⇒ not passed. +public sealed record SqlWriteSpec( + SqlWriteMode Mode, + bool StampTimestamp = false, + string? Procedure = null, + string? ValueParameter = null, + string? KeyParameter = null, + string? TimestampParameter = null) +{ + /// The default when a writable tag carries no write block: today's single-row UPDATE. + public static SqlWriteSpec Default { get; } = new(SqlWriteMode.Update); +} + +/// +/// Which s each provider serves. One table, read by the dialects (runtime), the deploy +/// gate (SqlTagDeployChecks) and the docs, so what deploy accepts is what the driver serves. +/// Odbc is Update / Insert only. The ANSI profile has no upsert every ODBC backend accepts +/// (MERGE is missing or partial on several), and positional ? 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. +/// +public static class SqlWriteSupport +{ + /// True when serves . + /// The driver's provider. + /// The tag's write mode. + /// Whether the driver will execute the mode on that provider. + 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, + }; +} + +/// +/// Parses and validates a tag's write block against the tag it belongs to. Used by +/// (a failure refuses the whole tag) and — through +/// — by the deploy gate and the AdminUI, so all three +/// share one verdict and one explanation. +/// +public static partial class SqlWriteSpecParser +{ + /// The keys a write block may carry. Anything else is refused — a typo in a write permission is + /// the last place to guess. + private static readonly string[] KnownKeys = + ["mode", "stampTimestamp", "procedure", "valueParam", "keyParam", "timestampParam"]; + + private static readonly string[] ProcedureOnlyKeys = ["procedure", "valueParam", "keyParam", "timestampParam"]; + + /// The default procedure parameter the value binds to. + public const string DefaultValueParameter = "value"; + + /// + /// Parses for (whose own fields are already parsed). + /// + /// The write property's value. + /// The tag definition the block belongs to. + /// The spec, when this returns . + /// An operator-facing reason, when this returns . + /// Whether the block is valid for the tag. + 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())}", 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; + } + + /// The column a stampTimestamp write sets: timestampColumn, else a WideRow's topByTimestamp. + /// The tag. + /// The column, or null when the tag has none. + 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; + } + + /// The tag's key value — what a procedure's keyParam binds: keyValue, whereValue or rowKey.value. + /// The tag. + /// The key, or null when the model names none. + 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, + }; + } + + /// + /// Reads a procedure parameter name: absent ⇒ ; otherwise a plain identifier + /// (letter or underscore, then letters, digits, underscores; at most 128 characters), one leading @ or + /// : tolerated and stripped. The name reaches SQL text unquoted (named notation), so this charset is + /// the injection boundary for it. + /// + 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(); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs index e1069372..0fbe6471 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/ISqlDialect.cs @@ -243,4 +243,65 @@ public interface ISqlDialect /// The bare SQL type family name; matched case-insensitively. /// The mapped driver data type, or when unrecognised. DriverDataType MapColumnType(string sqlDataType); + + // ---- writes (lmxopcua#781 item 2) ---- + + /// + /// Whether this dialect executes . Defaults to the shared + /// table for — the same table the deploy gate checks, so + /// a mode deploy accepts is a mode the driver serves. + /// + /// The tag's write mode. + /// Whether a write in that mode is executed. + bool SupportsWriteMode(SqlWriteMode mode) => SqlWriteSupport.IsSupported(Provider, mode); + + /// + /// 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. + /// + /// The quoted, catalog-canonical target table. + /// The quoted key column. + /// The key's parameter marker. + /// The quoted columns to set and their markers — the value first, then any stamp. + /// The statement, or null when the dialect cannot upsert. + string? UpsertSql( + string quotedTable, string quotedKeyColumn, string keyMarker, + IReadOnlyList<(string QuotedColumn, string Marker)> assignments) => null; + + /// + /// The most rows an upsert may report touching for one key. 1 everywhere except MySQL, whose + /// ON DUPLICATE KEY UPDATE reports 2 for "updated an existing row". + /// + int UpsertMaxRowsAffected => 1; + + /// + /// The statement that calls one stored procedure, or null when this dialect cannot. Arguments are passed by + /// name (EXEC p @a = @v, CALL p(a => @v), BEGIN p(a => :p_v); END;) 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. + /// + /// The quoted, catalog-canonical procedure name. + /// Procedure parameter name → marker, in order. + /// The statement, or null. + string? ProcedureCallSql(string quotedProcedure, IReadOnlyList<(string Parameter, string Marker)> arguments) + => null; + + /// + /// Catalog query listing the stored procedures in one schema (column ROUTINE_NAME, bound like + /// under the base name schema), or null when this dialect has no routine + /// catalog — the catalog gate then refuses every Procedure-mode tag rather than quote an unvalidated name. + /// + string? ListRoutinesSql => null; + + /// + /// 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 + /// when the provider reports one. + /// + /// What the write threw. + /// The failure class. + SqlWriteFailure ClassifyWriteFailure(Exception exception) + => SqlWriteFailures.FromSqlState((exception as DbException)?.SqlState); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/MySqlDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/MySqlDialect.cs index 81e777b7..f1b08618 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/MySqlDialect.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/MySqlDialect.cs @@ -177,4 +177,48 @@ public sealed class MySqlDialect : ISqlDialect _ => DriverDataType.String, }; } + + // ---- writes (lmxopcua#781 item 2) ---- + + /// + // 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))); + + /// + // 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; + + /// + // 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)) + ")"; + + /// + 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"; + + /// + // 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), + }; + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OdbcDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OdbcDialect.cs index 5f381c2f..6c00b582 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OdbcDialect.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OdbcDialect.cs @@ -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). + + /// + // 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; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OracleDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OracleDialect.cs index 1789959f..f298989d 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OracleDialect.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/OracleDialect.cs @@ -325,4 +325,50 @@ public sealed class OracleDialect : ISqlDialect _ => DriverDataType.String, }; } + + // ---- writes (lmxopcua#781 item 2) ---- + + /// + // 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)), ")"); + + /// + // 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;"; + + /// + // 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"; + + /// + // 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, + }; + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs index 6809b328..2de193c8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/PostgresDialect.cs @@ -267,4 +267,35 @@ public sealed class PostgresDialect : ISqlDialect _ => DriverDataType.String, }; } + + // ---- writes (lmxopcua#781 item 2) ---- + + /// + // 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))); + + /// + // 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)) + ")"; + + /// + // 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. } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlServerDialect.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlServerDialect.cs index 61057407..43fee530 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlServerDialect.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlServerDialect.cs @@ -222,4 +222,52 @@ public sealed class SqlServerDialect : ISqlDialect _ => DriverDataType.String, }; } + + // ---- writes (lmxopcua#781 item 2) ---- + + /// + // 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)), ");"); + + /// + // 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)); + + /// + // 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"; + + /// + // 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, + }; + } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlWriteFailure.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlWriteFailure.cs new file mode 100644 index 00000000..fe629ab6 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Dialect/SqlWriteFailure.cs @@ -0,0 +1,33 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Sql; + +/// Why a write's statement failed, as far as the provider's error says (lmxopcua#781 item 2). +public enum SqlWriteFailure +{ + /// Anything else — permission denied, a missing object, a broken connection. BadCommunicationError. + Other, + + /// A unique / primary-key violation (SQLSTATE 23505, SQL Server 2627/2601, MySQL 1062, ORA-00001). BadEntryExists. + UniqueViolation, + + /// Another integrity constraint — NOT NULL, foreign key, CHECK (SQLSTATE class 23). BadOutOfRange. + IntegrityViolation, + + /// A data exception — overflow, truncation, an unconvertible value (SQLSTATE class 22). BadOutOfRange. + DataException, +} + +/// SQLSTATE-based classification shared by the dialects. +public static class SqlWriteFailures +{ + /// Classifies an ANSI SQLSTATE: 23505 unique, class 23 integrity, class 22 data; anything else (or null) other. + /// The five-character SQLSTATE, or null. + /// The failure class. + 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; + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalog.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalog.cs index f69efa4d..63742636 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalog.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalog.cs @@ -23,17 +23,23 @@ public sealed class SqlCatalog /// Canonical schema.table → that relation's canonical column names. private readonly IReadOnlyDictionary> _columnsByTable; + /// Stored procedures per canonical schema — loaded only for schemas a Procedure-mode tag names (lmxopcua#781). + private readonly IReadOnlyDictionary> _routinesBySchema; + /// Constructs a catalog snapshot. /// The schema an unqualified object name resolves to for this connection. /// Every schema the connection can see. /// Canonical schema → its tables/views. /// Canonical schema.table → its columns. + /// Canonical schema → its stored procedures; null ⇒ none loaded. internal SqlCatalog( string defaultSchema, IReadOnlyList schemas, IReadOnlyDictionary> tablesBySchema, - IReadOnlyDictionary> columnsByTable) + IReadOnlyDictionary> columnsByTable, + IReadOnlyDictionary>? routinesBySchema = null) { + _routinesBySchema = routinesBySchema ?? new Dictionary>(StringComparer.Ordinal); DefaultSchema = defaultSchema; _schemas = schemas; _tablesBySchema = tablesBySchema; @@ -93,6 +99,21 @@ public sealed class SqlCatalog && TryMatch(columns, authored, out canonical); } + /// + /// 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. + /// + /// The schema, as returned it. + /// The authored procedure name. + /// The catalog spelling, when this returns true. + /// True when exactly one procedure matches. + 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); + } + /// The dictionary key for a canonical relation. /// The canonical schema. /// The canonical table. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogGate.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogGate.cs index ec8f34aa..01834d15 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogGate.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogGate.cs @@ -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; } + /// The rejection for a write block's procedure. + internal const string ProcedureField = "Write.Procedure"; + + /// + /// 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. + /// + 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; + } + /// Splits and resolves the authored table to a canonical schema + relation. private static bool TryResolveRelation( SqlTagDefinition definition, diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogLoader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogLoader.cs index cc82aced..6bfc6563 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogLoader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlCatalogLoader.cs @@ -37,6 +37,8 @@ internal static class SqlCatalogLoader private const string ColumnNameColumn = "COLUMN_NAME"; /// Base name of the schema parameter every dialect's catalog SQL carries. + private const string RoutineNameColumn = "ROUTINE_NAME"; + private const string SchemaParameter = "schema"; /// Base name of the table parameter carries. @@ -50,6 +52,7 @@ internal static class SqlCatalogLoader /// The distinct authored table strings, as written in the blobs. /// Per-command server-side backstop. /// Cancellation token for the operation. + /// Procedure-mode write targets (lmxopcua#781) whose schemas' procedures to list. /// The loaded catalog. /// The connection can see no schemas at all. internal static async Task LoadAsync( @@ -57,7 +60,8 @@ internal static class SqlCatalogLoader ISqlDialect dialect, IReadOnlyCollection authoredTables, TimeSpan commandTimeout, - CancellationToken cancellationToken) + CancellationToken cancellationToken, + IReadOnlyCollection? 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>(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); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs index 39f0fe3c..207ed352 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs @@ -15,11 +15,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql; /// connectivity, and the subscription overlay, while /// owns the read path. /// Writes are opt-in at two levels and default off (design §3.4, lmxopcua#753). A write needs -/// the driver's allowWrites, the tag's own "writable": true, 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 UPDATE over the tag's -/// catalog-validated identifiers (), run in a transaction that commits only when -/// exactly one row was touched. With allowWrites off every write is +/// the driver's allowWrites, the tag's own "writable": true, and a target its write 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 (), 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 allowWrites off every write is /// and every variable discovers as /// . /// Health is classified here, because nothing below does it. 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. /// 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); + + /// + /// 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. + /// + 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), + }; /// - // 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> WriteAsync( IReadOnlyList 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); } } - /// Opens a connection and runs one planned UPDATE in a transaction that commits only on one row. + /// The status a classified write failure publishes (lmxopcua#781; documented in docs/drivers/Sql.md). + internal static uint MapWriteFailure(SqlWriteFailure failure) => failure switch + { + SqlWriteFailure.UniqueViolation => SqlStatusCodes.BadEntryExists, + SqlWriteFailure.IntegrityViolation or SqlWriteFailure.DataException => SqlStatusCodes.BadOutOfRange, + _ => SqlStatusCodes.BadCommunicationError, + }; + + /// + /// Opens a connection and runs one planned write in a transaction. A table write commits only when it touched + /// between one and rows; a procedure call commits whenever it did + /// not throw (its row count is the procedure's own business). + /// private async Task 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 /// Opens one connection and reads the catalog slice the authored tables need. private async Task LoadCatalogAsync( - IReadOnlyCollection authoredTables, CancellationToken cancellationToken) + IReadOnlyCollection authoredTables, IReadOnlyCollection 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); } } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs index 6e98dfc3..3e476637 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs @@ -1257,7 +1257,22 @@ public static class SqlStatusCodes /// public const uint BadNotWritable = 0x803B0000u; - /// The provider could not report how many rows a write touched, so it was rolled back. + /// + /// A write the database refused for a unique / primary-key violation (lmxopcua#781) — an Insert of a key that + /// exists, typically. (BadEntryExists, read off Opc.Ua.StatusCodes.) + /// + public const uint BadEntryExists = 0x809F0000u; + + /// + /// 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. (BadOutOfRange.) + /// + public const uint BadOutOfRange = 0x803C0000u; + + /// + /// 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). + /// public const uint BadNotSupported = 0x803D0000u; /// The quality-class mask — the top two bits of an OPC UA status code. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlWritePlanner.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlWritePlanner.cs index f5388678..5669c0e6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlWritePlanner.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlWritePlanner.cs @@ -4,94 +4,235 @@ using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts; namespace ZB.MOM.WW.OtOpcUa.Driver.Sql; /// -/// One compiled single-row write (lmxopcua#753): a parameterised UPDATE 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. /// /// The statement — dialect-quoted identifiers and parameter markers only. -/// The bind names, in marker order (positional providers bind in this order). +/// +/// 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). +/// /// The bound values, aligned with . +/// The statement shape — decides how the executor reads the affected-row count. +/// +/// The most rows the statement may report for a success: 1, or 2 for a MySQL upsert. Ignored for +/// , whose row count is the procedure's business. +/// public sealed record SqlWritePlan( string SqlText, IReadOnlyList ParameterBindNames, - IReadOnlyList Parameters); + IReadOnlyList Parameters, + SqlWriteMode Mode = SqlWriteMode.Update, + int MaxRowsAffected = 1); /// -/// Turns one OPC UA write into a parameterised single-row UPDATE (design §3.4, lmxopcua#753) — pure, no -/// I/O. The write surface is deliberately narrow: no authored SQL, no INSERT/UPSERT/procedure, only -/// UPDATE <table> SET <value column> = @v WHERE <key column> = @k 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 write block picks the shape; there is no authored write SQL: /// -/// : SET valueColumn = @v WHERE keyColumn = @k. -/// with a where-pair: SET columnName = @v WHERE whereColumn = @w. -/// by topByTimestamp, and : -/// refused () — "the newest row" moves under a concurrent -/// insert, and a named query's projection names no table to update. +/// (the default): UPDATE t SET value [, stamp] WHERE key = @k +/// on a KeyValue tag or a WideRow where-pair. +/// : INSERT INTO t (key, value [, stamp]) VALUES (…) — no key +/// column for a WideRow topByTimestamp tag (an append). +/// : the dialect's single-statement upsert +/// (). +/// : the dialect's call of the catalog-validated procedure +/// () with the value, and optionally the key and write time, bound. /// -/// Injection boundary. Every identifier is one the catalog gate resolved and substituted (the -/// driver plans writes only for polled, i.e. gate-accepted, definitions) and reaches the text through -/// ; the written value and the key are bound parameters. Nothing an -/// OPC UA client sends reaches the command text. +/// Injection boundary. Every identifier is one the catalog gate resolved and substituted (the driver +/// plans writes only for gate-accepted definitions) and reaches the text through +/// ; 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. /// public static class SqlWritePlanner { private const string ValueBaseName = "v"; private const string KeyBaseName = "k"; + private const string TimestampBaseName = "t"; - /// - /// Plans a write of to , or explains why it is refused. - /// + /// Plans a write of to , stamping the current UTC time. /// The catalog-validated definition. /// The value the client wrote. - /// Quoting, markers and bind names. + /// Quoting, markers, bind names and the dialect-specific statements. /// The plan, when this returns . /// , or the status the write publishes. public static uint TryPlan(SqlTagDefinition tag, object? value, ISqlDialect dialect, out SqlWritePlan? plan) + => TryPlan(tag, value, dialect, DateTime.UtcNow, out plan); + + /// + /// Plans a write of to , or explains why it is refused. + /// : the model has no target for the mode (a Query tag outside + /// Procedure, a topByTimestamp row for Update/Upsert). : the + /// dialect does not serve the mode. : the value will not coerce. + /// + /// The catalog-validated definition. + /// The value the client wrote. + /// Quoting, markers, bind names and the dialect-specific statements. + /// The write time bound to a stamp column or timestampParam (UTC). + /// The plan, when this returns . + /// , or the status the write publishes. + 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]); + /// Update / Insert / Upsert against the tag's own table. + 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(); + var markers = new List(); + 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; + } + + /// A call of the tag's catalog-validated procedure: value, then key, then write time — each when bound. + 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; } - /// Quotes each part of a (catalog-canonical) schema.table name. + /// Quotes each part of a (catalog-canonical) schema.name. private static string QuoteQualifiedName(string name, ISqlDialect dialect) => string.Join('.', name.Split('.').Select(dialect.QuoteIdentifier)); private static bool Blank(string? value) => string.IsNullOrWhiteSpace(value); + + /// The ordered (bind name, value) list a plan carries. + private sealed class Binds(ISqlDialect dialect) + { + public List Names { get; } = []; + + public List Values { get; } = []; + + public void Add(string baseName, object? value) + { + Names.Add(dialect.ParameterBindName(baseName)); + Values.Add(value); + } + } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorSqlWriteBlockTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorSqlWriteBlockTests.cs new file mode 100644 index 00000000..160bb70a --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Configuration.Tests/DraftValidatorSqlWriteBlockTests.cs @@ -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; + +/// +/// lmxopcua#781 item 2: a Sql tag's write block is checked at deploy by the driver's own parser +/// (SqlWriteBlockInvalid), and its mode against the instance's provider (SqlWriteModeUnsupported). +/// Both always block. +/// +[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); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/mysql/seed-mysql.sql b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/mysql/seed-mysql.sql index fd13d6b7..871cc043 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/mysql/seed-mysql.sql +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/mysql/seed-mysql.sql @@ -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; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/odbc/run-odbc-gate.sh b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/odbc/run-odbc-gate.sh index 8e279839..df55c38f 100755 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/odbc/run-odbc-gate.sh +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/odbc/run-odbc-gate.sh @@ -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" diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/oracle/seed-oracle.sql b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/oracle/seed-oracle.sql index 28d92da1..1c0a9f15 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/oracle/seed-oracle.sql +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/oracle/seed-oracle.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; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql index 59ce06da..26e8fc13 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/Docker/postgres/seed-pg.sql @@ -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; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/OdbcReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/OdbcReadTests.cs index aea984ce..6c80ed05 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/OdbcReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/OdbcReadTests.cs @@ -111,6 +111,43 @@ public sealed class OdbcReadTests columns.Select(c => c.DisplayName).ShouldBe(["tag_name", "num_value", "sample_ts"]); } + /// + /// lmxopcua#781: the two write modes Odbc serves, through positional ? 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. + /// + [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 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 } /// Builds the driver through the production factory with provider: Odbc. - private static async Task StartAsync(string? queriesJson, params RawTagEntry[] tags) + private static Task StartAsync(string? queriesJson, params RawTagEntry[] tags) + => StartAsync(queriesJson, allowWrites: false, tags); + + /// Builds the driver through the production factory with provider: Odbc. + private static async Task 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); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlPollServerFixture.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlPollServerFixture.cs index 16c0ed61..33c71e6e 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlPollServerFixture.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlPollServerFixture.cs @@ -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 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 } /// Runs one non-query batch. + private static IEnumerable 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(); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlWriteModesLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlWriteModesLiveTests.cs new file mode 100644 index 00000000..85763640 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.IntegrationTests/SqlWriteModesLiveTests.cs @@ -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; + +/// +/// The authored write 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. +/// Every seed carries the same shape (Docker/*/seed-*.sql, and +/// for SQL Server): write_targets(tag_name PK, sp_value NOT NULL, stamp_ts) holding one Present +/// row, and set_target(p_value, p_key, p_ts), which journals its arguments into proc_calls. Each +/// test writes a key of its own, so the suite needs no cleanup grant and tolerates re-runs. +/// +public abstract class SqlWriteModesLiveSuite +{ + /// Why the backend is unavailable, or null. + protected abstract string? SkipReason { get; } + + /// The suite login's connection string. + protected abstract string ConnectionString { get; } + + /// The backend's dialect. + protected abstract ISqlDialect Dialect { get; } + + /// The schema the seed lives in, as authored (the catalog gate folds its case). + protected abstract string Schema { get; } + + /// The procedure as authored — deliberately in the case the backend does NOT store, so the gate's spelling substitution is exercised. + 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 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 WriteAsync(SqlDriver driver, string reference, object? value) + => (await ((IWritable)driver).WriteAsync([new WriteRequest(reference, value)], TestContext.Current.CancellationToken)) + .ShouldHaveSingleItem().StatusCode; + + private async Task CountAsync(string table, string key) + => Convert.ToInt64(await ScalarAsync($"SELECT COUNT(*) FROM {{0}} WHERE tag_name = '{key}'", table), + CultureInfo.InvariantCulture); + + /// Runs a scalar as the suite login; {0} is replaced by the schema-qualified table. + private async Task 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; + } + } + } +} + +/// SQL Server leg — the seeding fixture on the shared central server (its own database). +[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"; +} + +/// PostgreSQL leg — the granted-not-owning login; upper-case authoring folds to the stored lower case. +[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"; +} + +/// MySQL leg — positional CALL, ON DUPLICATE KEY UPDATE's 1-or-2 row count. +[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"; +} + +/// Oracle leg — MERGE from DUAL, an anonymous PL/SQL block; lower-case authoring folds to the stored upper case. +[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"; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlWriteModeTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlWriteModeTests.cs new file mode 100644 index 00000000..eb33e7c1 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlWriteModeTests.cs @@ -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; + +/// +/// The authored write 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. +/// +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); + + /// + /// 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. + /// + [Fact] + public void EveryRealDialect_buildsAStatement_exactlyForTheModesTheSharedTableServes() + { + foreach (var provider in Enum.GetValues()) + { + 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>(StringComparer.Ordinal) { ["dbo"] = ["sp"] }, + new Dictionary>(StringComparer.Ordinal) { ["dbo.sp"] = ["tag", "val"] }, + new Dictionary>(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()) + 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); + } + + /// The same absent key under the default Update is BadNoData and inserts nothing — the back-compat pin. + [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 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 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 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) { } + } + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqliteDialect.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqliteDialect.cs index 20dded60..d9ba01f4 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqliteDialect.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqliteDialect.cs @@ -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 ---- + + /// + // 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; + + /// + 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))); + + /// + // 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; } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/SqlTagConfigModelTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/SqlTagConfigModelTests.cs index 6a5b655a..e72ec497 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/SqlTagConfigModelTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Uns/SqlTagConfigModelTests.cs @@ -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'"); + } }