diff --git a/CLAUDE.md b/CLAUDE.md index fc86bc04..9447335c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1350,7 +1350,7 @@ historian tagname, which is the tag's driver `FullName`. ### Pluggable backends, SQL Server and HistoryUpdate (2026-10) - **Selection:** `ServerHistorian:Backend` (`Gateway` default | `SqlServer` | `QuestDb`; unknown fails host start). `HistorianBackendRegistry` → `IHistorianBackendProvider` builds all five seams; the gateway keys are only required for `Gateway`. `SqlServer` uses `ServerHistorian:SqlServer:{ConnectionString (required), Schema=hist, CommandTimeout=30s, AutoCreateSchema=true}`. -- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (10, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted seven (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM` — plus the **duration-based** PercentGood/PercentBad, #680) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`, percentages included). ⚠️ **#680 status rules (client-visible):** TimeAverage/Total are Uncertain if any part of the interval uses an Uncertain value, is bridged across a Bad sample, is extrapolated past the last sample (⇒ a live tag's latest interval is Uncertain) or is not covered; TimeAverage2/Total2 keep the SDK time-based rule (Uncertain time counts as Good) but never extrapolate past the last sample — that interval is Partial, later ones `BadNoData` (a report-by-exception tag with nothing after its last sample reads `BadNoData` until its next sample arrives). `docs/Historian.md` §"Time-weighted aggregates". ⚠️ **#680 per-sample (client-visible):** `Count` counts **Good** samples only (value-based status; an all-Bad interval is generic Bad, value null) and an empty interval **inside the tag's data range** is `0`/Good (outside it still `BadNoData`); `Start`/`End` return the **raw** first/last sample of any quality — its own timestamp and status, value null when Bad; `Delta` is Uncertain only when the interval's first or last sample is unusable; ⚠️ #684: `Average`/`Minimum`/`Maximum`/`Range`/`*ActualTime` use **Good samples only** (was Good + Uncertain; an Uncertain-only interval is `BadNoData`), Min/Max-family status Uncertain on a Bad sample, an Uncertain value beyond the Good extreme (Part 13 — the SDK ignores that) or an Uncertain non-numeric sample, and only a **Good** String/DateTime sample refuses them (`BadAggregateInvalidInputs`; Delta still refuses any usable one); and every per-sample aggregate but `Average` sets **end-of-data Partial** on the interval holding the last sample (forward incl. its earlier edge, reverse strictly inside), never on a Bad result. The data-range lookup is a seek on SQL Server but a partition scan on QuestDB (unindexed `tag` SYMBOL). `docs/Historian.md` §"Aggregates". Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. +- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (10, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted seven (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM` — plus the **duration-based** PercentGood/PercentBad, #680) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; reverse `Interpolative` anchored at the later time; window capped by `ServerHistorian:MaxTimeWeightedRawRows` ⇒ `BadTooManyOperations`, percentages included). ⚠️ **#680 status rules (client-visible):** TimeAverage/Total are Uncertain if any part of the interval uses an Uncertain value, is bridged across a Bad sample, is extrapolated past the last sample (⇒ a live tag's latest interval is Uncertain) or is not covered; TimeAverage2/Total2 keep the SDK time-based rule (Uncertain time counts as Good) but never extrapolate past the last sample — that interval is Partial, later ones `BadNoData` (a report-by-exception tag with nothing after its last sample reads `BadNoData` until its next sample arrives). `docs/Historian.md` §"Time-weighted aggregates". ⚠️ **#680 per-sample (client-visible):** `Count` counts **Good** samples only (value-based status; an all-Bad interval is generic Bad, value null) and an empty interval **inside the tag's data range** is `0`/Good (outside it still `BadNoData`); `Start`/`End` return the **raw** first/last sample of any quality — its own timestamp and status, value null when Bad; `Delta` is Uncertain only when the interval's first or last sample is unusable; ⚠️ #684: `Average`/`Minimum`/`Maximum`/`Range`/`*ActualTime` use **Good samples only** (was Good + Uncertain; an Uncertain-only interval is `BadNoData`), Min/Max-family status Uncertain on a Bad sample, an Uncertain value beyond the Good extreme (Part 13 — the SDK ignores that) or an Uncertain non-numeric sample, and only a **Good** String/DateTime sample refuses them (`BadAggregateInvalidInputs`; Delta still refuses any usable one); and every per-sample aggregate but `Average` sets **end-of-data Partial** on the interval holding the last sample (forward incl. its earlier edge, reverse strictly inside), never on a Bad result. The data-range lookup is a key seek on SQL Server and a `tag` symbol-index seek on QuestDB (#683 — `{p}sample.tag` is `SYMBOL INDEX`; the bootstrapper ADDs it to a pre-#683 table, idempotent), read BEFORE the bucket statement so a racing append can never yield a Good `Count` of 0 (a concurrent HistoryUpdate delete is the mirror case and still can, transiently). `docs/Historian.md` §"Aggregates". Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. **Retention (#667):** opt-in `SqlServer:RetentionDays` (+ `PurgeInterval` 1h, `PurgeBatchSize` 4000) -> `HistorianRetentionService` purges old `hist.Sample` (per tag) / `hist.Event` in `DELETE TOP(n)` batches, Primary-gated on `IHistorizationRoleView`, one purger per DB via `sp_getapplock`; counter `otopcua.historian.retention.purged{table}`. Optional `IHistorianBackendProvider.CreateRetention` (default null). `docs/Historian.md` §Retention. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - ⚠️ **Reverse processed reads are Part 13/SDK-anchored (#677, client-visible):** every aggregate on SqlServer/QuestDB now anchors a reverse read at its later time — window `(end, start]`, partial interval earliest, each stamped with its later edge (was: earlier-anchored, stamped earlier edge); on forward reads Partial also marks the interval the data starts inside (not Average — the SDK's rule, forward-only); a non-default client `AggregateConfiguration` ⇒ `BadAggregateConfigurationRejected`; Gateway provisioning sends `IntegralDivisor = 1`. SDK-parity tests: `SdkAggregateParityTests`. `docs/Historian.md` §Aggregates. - **Advertised aggregates (#682):** `IHistorianDataSource.SupportedAggregates` (Gateway 4 via `GatewayHistorianDataSource.ServedAggregates`, SqlServer/QuestDB 17 via `AggregateFamilies.Served`, Null **none**; a null set ⇒ all) is the one authority for `HistoryServerCapabilities/AggregateFunctions` (i=11201 — only its folder references are trimmed) and the up-front `BadAggregateNotSupported` in HistoryRead-Processed. ⚠️ `ServerCapabilities/AggregateFunctions` (i=2997) is deliberately NOT trimmed: it keeps the SDK's 37 because live `AggregateFilter` subscriptions still compute them all. ⚠️ Historian disabled ⇒ Processed is now `BadAggregateNotSupported`, not `GoodNoData`. `docs/Historian.md` §Advertised aggregates. diff --git a/docs/Historian.md b/docs/Historian.md index e7ab23ec..71eda810 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -1124,9 +1124,27 @@ Data continuing past the window sets nothing. Two deliberate limits, kept: the p **Data-range queries.** The start/end-of-data rules and `Count`'s in-range empty interval need the tag's **data range**: its earliest stored sample (any quality) at or before the window's later edge and its latest at or after the earlier edge. On **SQL Server** these are one extra result set of two `MIN`/`MAX` subqueries, each a single seek on the `(TagId, SourceTimeUtc)` key. On **QuestDB** they are -two `LIMIT 1` statements, but `tag` is an **unindexed `SYMBOL`**, so each is a **partition scan**, not a seek: measured 1–3 ms for a -live tag, and for a stopped tag read over an old window 6–8 ms warm / ~32 ms cold on a 52 M-row table — the cost grows with the table -(bounded by the TTL), not with the window. They run only when the answer can change a result +two `LIMIT 1` statements that **seek the `tag` symbol index** (#683) partition by partition. Before #683 `tag` was unindexed and +each was a partition scan whose cost grew with the table, not the window — worst for a stopped tag read over an old window (`LatestSample` +scans every later partition) and a late-starting tag (`EarliestSample` scans from the table start). Measured on 53.6 M rows / 120 tags / +61 daily partitions (QuestDB 10.0.1, HTTP round-trip included, median of 7): + +| Data-range lookup | unindexed warm / cold | indexed warm / cold | +|---|---|---| +| live tag (either lookup) | 1.5–3 ms / 2–14 ms | 1.1–1.4 ms / 2.5–3.5 ms | +| stopped tag, window inside its data (`LatestSample`) | 11 ms / 11 ms | 1.5 ms / 3.5 ms | +| stopped tag, window after it stopped (`LatestSample`, no row) | 7 ms / 9 ms | 1.0 ms / 1.9 ms | +| late-starting tag (`EarliestSample` from table start) | 7–7.5 ms / 7–10 ms | 1.2 ms / 2–3 ms | + +With the index an absent side still visits every partition on that side, but as one index-key lookup each rather than a column scan, so +it no longer grows with the rows per partition (the tag count); it still grows with the partition count (bounded by the TTL). Alternatives +measured and rejected: bounded existence probes ("any sample before the window start / after its end", `LIMIT 1` from the window edge) +are cheap only when data *exists* on that side — an absent side is still a full scan, the very case the issue is about; `LATEST ON ts +PARTITION BY tag` was the slowest unindexed form (19–32 ms) and no faster than the plain seek indexed; and a per-tag first/last-write side +table cannot be kept correct INSERT-only — a TTL partition drop or a HistoryUpdate delete moves the first/last sample with no write to hook. +They are read **before** the bucket statement (a separate table reader), so a sample appended in between reaches the buckets but not the +range — at worst a stale end-of-data Partial or a `BadNoData` where a `Count` of 0 belongs, never a Good `0` for an interval that has +samples (the result of the opposite order). SQL Server reads its range result set first for the same reason. They run only when the answer can change a result (`AggregateEvaluator.NeedsDataRange`): a per-sample read of any aggregate other than `Average`, in either direction (`Count` included); a per-sample `Average` read skips them. @@ -1452,11 +1470,34 @@ makes HistoryUpdate Replace work; the writer's own insert-if-absent semantics ar | Table | Key | Partitioning | |---|---|---| -| `{p}sample` (`ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts`) | `(ts, tag)` | `PARTITION BY DAY` (TTL when configured) | +| `{p}sample` (`ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts`; **`tag` is a `SYMBOL INDEX`**, #683) | `(ts, tag)` | `PARTITION BY DAY` (TTL when configured) | | `{p}event` (one-for-one with `hist.Event`) | `(ts, event_id)` | `PARTITION BY DAY` (TTL when configured) | | `{p}tag` (`ts, name, data_type, eng_unit, description, updated_utc`) | `(ts, name)` with `ts` fixed at `1970-01-01` — an upsert by name | `PARTITION BY YEAR`, never expires | | `{p}sample_modified` (`ts` = the sample's source time, `tag, modified_id, modification_ts, update_type, principal, api_key_id, kind, num_value, int_value, str_value, status, server_ts`) (#668) | none — append-only, no DEDUP; `modified_id` (process-monotonic ≈ UTC ticks) orders a timestamp's trail | `PARTITION BY DAY` (TTL when configured). A NULL `status` marks "no prior value" (QuestDB `BYTE` has no NULL, so `kind` is 0) | +**The `tag` index (#683).** Every read filters on one tag; without a symbol index QuestDB answers `tag = x` by scanning the tag +column of every partition in the time range, which made the processed-read [data-range queries](#aggregates-processed-reads) of a stopped or late-starting +tag scan every partition on one side of the window. A table created before #683 has no index: with `AutoCreateSchema=true` the +bootstrapper adds it on first use (`ALTER TABLE {p}sample ALTER COLUMN tag ADD INDEX`, a WAL transaction; idempotent — re-sending it on an +indexed column is harmless, so a pair booting together is safe — and a boot that finds the column indexed sends nothing). Building it is a +one-off WAL apply, measured **~3 s for 68 M rows / 63 daily partitions**, during which writes queue and reads keep scanning. With +`AutoCreateSchema=false` a missing index only logs a Warning naming that statement — reads stay correct, just slower. With +`AutoCreateSchema=true` an ALTER refused up front (e.g. no DDL rights) is likewise only a Warning — the bootstrap still succeeds and the +next boot retries; a failure at **WAL-apply** time is not observed by the bootstrapper (QuestDB would suspend the table, visible in +`wal_tables()`) — the same blind spot as the TTL ALTER. Cost: ~12 % more disk (0.5 GB on a 4.2 GB 53 M-row table); no measurable change +in live-append, O3 back-fill or bulk-load apply time beyond run-to-run noise (a first bulk load into an empty table measured ~40 % +slower). Out-of-order writes (back-fill, HistoryUpdate, outbox redelivery) rewrite the index files of every partition they touch along +with the column files, so their apply cost carries the index too. On a 120-tag table the per-sample / raw / time-weighted window queries +were unchanged or slightly faster (the tag filter becomes an index scan), but **a tag holding a large share of the table can be ~20 % +slower on long windows**: the index scan is single-threaded where the unindexed filter is a multi-threaded JIT scan (a review +measurement: 4 tags / 30 M rows, 245–294 ms indexed vs 195–218 ms unindexed; with 1000 tags, 9 ms vs 13 ms — the index wins as the +tag's share of each partition shrinks). + +**Deliberately not indexed:** `{p}sample_modified.tag` — read only by `IsReadModified` HistoryRead over a window, an append-only trail a +tiny fraction of the sample table's size, with no "nearest row outside the window" lookup to make it scan partitions; and +`{p}event.source` — event reads are window-bounded (`ts` range, optionally `source = x`) and never seek beyond the window, and the event table is far +smaller than the sample table. Neither has the unbounded-scan shape #683 fixes, so neither pays the index's disk and O3 cost. + `status` holds the uint OPC UA status code **unsigned** in a `LONG` (so "Bad" is `status >= 2147483648`, not `< 0`); `int_value` also holds Booleans as 0/1; `kind` is the `HistorianValueKind` in a `BYTE`. Tag names are limited to 400 characters (longer is a per-tag failure, never truncated). `{p}tag` is the **only** "is this tag known" authority (`QuestDbTagCatalog`): reads never create a tag, Insert/Update and the @@ -1552,8 +1593,8 @@ Every query filters `NOT deleted`; every value goes through `StoredSample`, so t later time, later-edge-inclusive, partial bucket earliest — #677; interval 0 ⇒ one bucket, 100,000-bucket cap). On a per-sample read of any aggregate but Average, in either direction, two `LIMIT 1` statements fetch the tag's data range — its earliest live sample at or before the window's later edge and its latest at or after its earlier edge — for the start/end-of-data Partial bits and Count's in-range empty interval - (#677/#680). `tag` is an unindexed SYMBOL, so these are partition scans (a few ms on a live tag; tens of ms cold for a stopped tag on a - large table — see *Data-range queries* above). Start/End/Delta also return each bucket's first and last sample of any quality (`row_number()` over every row). + (#677/#680), read before the bucket statement. They seek the `tag` symbol index (#683; ~1–1.5 ms warm, 2–4 ms cold on a 53 M-row table + whatever the tag's history — see *Data-range queries* above). Start/End/Delta also return each bucket's first and last sample of any quality (`row_number()` over every row). Min/max times use `row_number()` windows with the earliest-timestamp tie-break. ⚠️ **Live-gate 8 F1 — counts and picks must come from one table version.** QuestDB opens a separate table reader per reference to a CTE, so the earlier Start/End/Delta form (counts CTE + self-joins) could count a row committed between two references without @@ -1561,8 +1602,37 @@ Every query filters `NOT deleted`; every value goes through `StoredSample`, so t query now reads the table **once**: one `GROUP BY` with `first()`/`last()` for the first/last sample of any quality and `arg_min`/`arg_max` keyed on the usable rows' time for the first/last usable one (~15 % faster than before; the usable picks carry no `str_value` — `arg_min` has no VARCHAR form, and only Delta reads them, never with a usable string in the bucket). On `SqlServer` - (READ COMMITTED gives no cross-reference guarantee either, though it was never observed there) Start/End/Delta was reshaped the same - way with a windowed pass + conditional `MAX` — 1.4–3× *faster* than the old five-reference form on a 1 M-row tag. Min/MaxActualTime + (READ COMMITTED gives no cross-reference guarantee either, though it was never observed there) Start/End/Delta was first reshaped + into a windowed pass + conditional `MAX`; **#683 replaced that** — its four `ROW_NUMBER` sorts spilled and it measured *slower* than + the five-reference form on long windows. It is now **aggregate, then join back on the key**: one `GROUP BY` computes the counts and each + bucket's four pick *times* (`MIN`/`MAX(SourceTimeUtc)`, any quality and usable), and each pick's columns are fetched by a seek on the + clustered key `(TagId, SourceTimeUtc)`. Output is identical (`EXCEPT` both ways empty on 6 M rows, forward and reverse); measured on + 6 M rows / 30 days, one tag (so a 1 d or 7 d window leaves up to 5.8 M rows of the tag's history outside it), average of 3 runs + over 2 rounds, plans compiled by the 30 d read and reused (sniffed) by the rest: + + | Window | windowed form (before) | join-back | + |---|---|---| + | 30 d @ 1 h (720 buckets) | 19.0–21.7 s | 3.4–3.5 s | + | 30 d @ 1 h, reverse | 24.7–30.3 s | 3.4 s | + | 7 d @ 1 d (7 buckets) | 5.5–6.0 s | 0.78–0.81 s | + | 7 d @ 1 min (10,080 buckets) | 2.1–2.3 s | 1.08–1.11 s | + | 1 d @ 1 h (24 buckets) | 0.55–0.60 s | 0.12–0.13 s | + | 1 d @ 1 h, reverse | 0.52–0.55 s | 0.11 s | + | 1 d @ 1 min (1440 buckets) | 0.29–0.32 s | 0.16 s | + + ⚠️ **Every pick join is bounded to the window** (`@start`/`@end`, the same predicate as the window CTE), not only keyed on the pick + time. Keyed alone, the optimizer may choose a hash/merge/adaptive join whose inner input is the tag's **whole history**, and + parameter sniffing then reuses that plan for every tag — a review measured 43,657 logical reads vs 361 for a 1 d @ 1 h read of a tag + with 3 M rows outside the window. Forcing that shape here, 1 d @ 1 h: **152,524** session logical reads unbounded vs **6,756** bounded + under `OPTION (HASH JOIN)` (6,655 under `MERGE JOIN`; the unforced plan reads 1,846 either way). `StartEndJoinBackTests` pins it: + a tag with 300 k rows either side of a 1 h window, forced hash and merge joins, < 1,000 reads (183 bounded, 8,055 with the bound + removed). Wider windows trade reads for time — 7 d @ 1 min reads 174 k pages (four key seeks per bucket) against the windowed form's + 46 k, yet runs in half the time. (The dev SQL container is slow in absolute terms — the ratios are the point.) + + ⚠️ The seeks are separate reads of the table within one statement, so a sample **deleted** by HistoryUpdate between the aggregate + and its seek leaves that pick NULL while the counts include it: the null-tolerant reader below answers that bucket's Start/End/Delta + `BadNoData`, transiently. A usable pick's join also re-checks usability (the exact `u` predicate), so a HistoryUpdate **Replace** + that turned the picked sample Bad/Empty in between reads the same way — never a Bad row's value under Delta. Min/MaxActualTime keep their counts + filtered-rank form on both backends: every single-scan variant measured 1.2–2.4× slower, and QuestDB's `arg_max` cannot express the earliest-time tie-break. So, on both backends, a row whose picks are absent anyway maps to "no such sample" — `BadNoData` for that interval, never an exception and never a value-less Good. `LiveWriteReadRaceStressTests` (QuestDB diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index 53f134a3..5bc333ea 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -450,7 +450,9 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// the SDK: window (end, start], buckets stepping back from start, the partial one earliest, each stamped /// with its later edge, returned latest first. One statement per node returns the statistics /// of each non-empty bucket (bucket index (cast(ts AS LONG)*10 − BucketBaseUnixTicks) / IntervalTicks, never - /// SAMPLE BY); a second fetches the tag's earliest sample for the start-of-data Partial bit; and + /// SAMPLE BY); before it, when , two tag-index seeks fetch the + /// tag's data range (earliest / latest live sample) for the start/end-of-data Partial bits and Count's in-range + /// empty interval; and /// produces the values and fills the empty buckets with BadNoData. The time-weighted family (#671) instead streams the window plus four bounding /// rows into , exactly as the SQL Server backend does. /// @@ -488,6 +490,20 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource _ => QuestDbAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed), }; await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false); + // The tag's data range (earliest + latest live sample): the start/end-of-data Partial rules (#677/#680 — + // any per-sample aggregate but Average, either direction) and Count's in-range empty interval (#680); skipped + // otherwise. Index seeks on tag (#683). Read BEFORE the bucket statement, which is a separate table reader: + // a sample appended in between (the common race — a live tag read up to now) then reaches the buckets but + // not the range — at worst a stale end-of-data Partial or a BadNoData where a 0 belongs — instead of a + // range covering a bucket the rows do not show, which reported a Good Count of 0 for an interval that has + // samples. (A concurrent HistoryUpdate delete is the mirror case and can still do that, transiently.) + if (AggregateEvaluator.NeedsDataRange(aggregate, plan)) + { + earliest = await ReadBoundAsync(conn, QuestDbAggregateQueries.EarliestSample(sample), fullReference, + plan.EndUtc, cancellationToken).ConfigureAwait(false); + latest = await ReadBoundAsync(conn, QuestDbAggregateQueries.LatestSample(sample), fullReference, + plan.StartUtc, cancellationToken).ConfigureAwait(false); + } await using (var cmd = _conn.CreateCommand(conn, sql)) { QuestDbConnection.BindLong(cmd, plan.BucketBaseUnixTicks); @@ -499,16 +515,6 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) rows.Add(ReadBucket(reader, family)); } - // The tag's data range (earliest + latest live sample): the start/end-of-data Partial rules (#677/#680 — - // any per-sample aggregate but Average, either direction) and Count's in-range empty interval (#680); skipped - // otherwise. - if (AggregateEvaluator.NeedsDataRange(aggregate, plan)) - { - earliest = await ReadBoundAsync(conn, QuestDbAggregateQueries.EarliestSample(sample), fullReference, - plan.EndUtc, cancellationToken).ConfigureAwait(false); - latest = await ReadBoundAsync(conn, QuestDbAggregateQueries.LatestSample(sample), fullReference, - plan.StartUtc, cancellationToken).ConfigureAwait(false); - } } var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, latest, now); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbSchemaBootstrapper.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbSchemaBootstrapper.cs index f610fac5..0dae5019 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbSchemaBootstrapper.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbSchemaBootstrapper.cs @@ -1,3 +1,4 @@ +using System.Globalization; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Npgsql; @@ -15,10 +16,17 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// TTL (RC3): tables() normalises units (7 days reads back as 1 WEEK) so values are compared /// in hours, and an ALTER becomes visible only after the WAL applies it, so the bootstrapper never re-reads to verify /// its own ALTER. A boot whose TTL already matches issues no ALTER. +/// Tag index (#683): a sample table created before the tag symbol index existed gains it with +/// (idempotent; a boot that finds it indexed issues no ALTER, and like the TTL +/// the ALTER is a WAL transaction the bootstrapper does not re-read). Building it is a one-off WAL apply — ~3 s for +/// 68M rows / 63 daily partitions measured — during which writes queue and reads keep scanning. A refused ALTER is +/// logged as a Warning, never fails the bootstrap (the index is optional); a failure at WAL-apply time would suspend +/// the table unseen, the same blind spot as the TTL ALTER. /// Failure: not latched. Within the last failure is /// re-thrown without touching QuestDB. /// AutoCreateSchema=false only verifies the tables exist (no TTL changes) and throws -/// otherwise. +/// otherwise; a missing tag index is only logged (a Warning naming +/// the ALTER to run), since reads stay correct without it. /// /// internal sealed class QuestDbSchemaBootstrapper @@ -63,10 +71,12 @@ internal sealed class QuestDbSchemaBootstrapper { await CreateTablesAsync(conn, ct).ConfigureAwait(false); await ReconcileTtlAsync(conn, ct).ConfigureAwait(false); + await EnsureTagIndexAsync(conn, apply: true, ct).ConfigureAwait(false); } else { await VerifyAsync(conn, ct).ConfigureAwait(false); + await EnsureTagIndexAsync(conn, apply: false, ct).ConfigureAwait(false); } _ready = true; _failure.Clear(); @@ -117,6 +127,44 @@ internal sealed class QuestDbSchemaBootstrapper } } + /// + /// #683: indexes the sample table's tag column when it is not (a table created before the index existed). + /// false (AutoCreateSchema off) only logs the statement to run. + /// + private async Task EnsureTagIndexAsync(NpgsqlConnection conn, bool apply, CancellationToken ct) + { + var table = _conn.SampleTable; + await using (var cmd = _conn.CreateCommand(conn, QuestDbTables.TagIndexedQuery(table))) + { + var indexed = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false); + if (indexed is null or DBNull || Convert.ToBoolean(indexed, CultureInfo.InvariantCulture)) return; + } + if (!apply) + { + _log.LogWarning("QuestDB historian table {Table} has no index on its tag column, so the data-range lookups of " + + "processed reads scan partitions instead of seeking (#683). AutoCreateSchema is false; run: {Statement}", + table, QuestDbTables.AddTagIndex(table)); + return; + } + // A WAL transaction: table_columns() keeps reporting false until it applies, so do NOT re-read to verify. + // The index is optional (reads are correct without it), so a refused ALTER (e.g. no DDL rights) must not fail the + // bootstrap: it is logged and retried on the next boot. A failure at WAL-APPLY time is not seen here (QuestDB would + // suspend the table) — the same blind spot as the TTL ALTER. + try + { + await using var alter = _conn.CreateCommand(conn, QuestDbTables.AddTagIndex(table)); + await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested) + { + _log.LogWarning("Could not add the tag index to QuestDB table {Table} ({Exception}: {Message}); data-range lookups " + + "of processed reads scan partitions instead of seeking (#683). Run: {Statement}", + table, ex.GetType().Name, ex.Message, QuestDbTables.AddTagIndex(table)); + return; + } + _log.LogInformation("Added the tag index to QuestDB table {Table} (#683).", table); + } + private async Task ReconcileTtlAsync(NpgsqlConnection conn, CancellationToken ct) { var wantHours = _options.RetentionDays is { } d ? d * 24L : 0L; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTables.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTables.cs index 595c10a9..5600905a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTables.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTables.cs @@ -39,9 +39,15 @@ internal static class QuestDbTables /// public static string[] Expiring(string prefix) => [Sample(prefix), Event(prefix), SampleModified(prefix)]; - /// CREATE TABLE IF NOT EXISTS for the sample table. + /// + /// CREATE TABLE IF NOT EXISTS for the sample table. tag carries a symbol (bitmap) index (#683): every + /// read filters on one tag, and without the index QuestDB answers a tag = x seek by scanning the tag column of + /// every partition in the time range — the data-range lookups ( / + /// LatestSample) of a stopped or late-starting tag then scan every partition on one side of the window. + /// A table created before #683 gains the index through . + /// public static string CreateSample(string prefix, int? retentionDays) => - $"CREATE TABLE IF NOT EXISTS {Sample(prefix)} (ts TIMESTAMP, tag SYMBOL, kind BYTE, num_value DOUBLE, " + + $"CREATE TABLE IF NOT EXISTS {Sample(prefix)} (ts TIMESTAMP, tag SYMBOL INDEX, kind BYTE, num_value DOUBLE, " + "int_value LONG, str_value VARCHAR, status LONG, deleted BOOLEAN, server_ts TIMESTAMP) " + $"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL DEDUP UPSERT KEYS(ts, tag)"; @@ -74,6 +80,21 @@ internal static class QuestDbTables [CreateSample(prefix, retentionDays), CreateEvent(prefix, retentionDays), CreateTag(prefix), CreateSampleModified(prefix, retentionDays)]; + /// + /// ALTER TABLE t ALTER COLUMN tag ADD INDEX — the #683 migration of a sample table created without the tag + /// index. A WAL transaction (visible once applied); re-issuing it on an indexed column is a harmless no-op (the table + /// is not suspended), so two nodes booting together may both send it. + /// + public static string AddTagIndex(string table) => $"ALTER TABLE {table} ALTER COLUMN tag ADD INDEX"; + + /// + /// Whether 's tag column is indexed: one row, one BOOLEAN column, no row when the + /// table or column is missing. The name is inlined (table_columns() takes a constant); it is always a + /// prefix-validated name built here. + /// + public static string TagIndexedQuery(string table) => + $"SELECT indexed FROM table_columns('{table}') WHERE \"column\" = 'tag'"; + /// ALTER TABLE t SET TTL n DAYS; null removes the TTL (SET TTL 0 DAYS). public static string AlterTtl(string table, int? retentionDays) => $"ALTER TABLE {table} SET TTL {(retentionDays ?? 0).ToString(CultureInfo.InvariantCulture)} DAYS"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs index d57120e5..c2d953cc 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs @@ -78,36 +78,59 @@ internal static class SqlAggregateQueries /// StatusCode) and 22..27 the last (same order) — present for every returned (non-empty) bucket. /// /// - /// One scan (live-gate 8 F1). The counts and all four picked samples come from one reference to s: - /// a windowed pass ranks each row within its bucket (rfa/rla over every sample, rfu/rlu - /// within its usable / not-usable partition) and the outer GROUP BY both counts and lifts each picked row's - /// columns with a conditional MAX (one row per bucket matches each pick). The previous form referenced the - /// CTE five times; SQL Server expands each reference into its own scan, and under READ COMMITTED nothing makes those - /// scans see the same rows, so a sample committed between two of them could be counted but not picked (QuestDB hit - /// exactly that). It is also fewer scans of the key range. + /// + /// Aggregate, then join back on the key (#683). One GROUP BY over the window computes the counts and + /// each bucket's four pick TIMES — MIN/MAX(SourceTimeUtc) over every sample (any quality) and over + /// the usable ones — and each pick's columns are then fetched by a seek on the clustered key + /// (TagId, SourceTimeUtc) (unique, so one row per pick; a NULL usable time joins nothing ⇒ NULL columns, + /// as before). Output is identical to the previous windowed form (one ROW_NUMBER pass + conditional + /// MAX, live-gate 8 F1) — EXCEPT both ways is empty on 6M rows, forward and reverse — without that + /// form's four window sorts, which spilled: on 6M rows / 30 days, 30 d @ 1 h ~3.4 s vs 18–32 s, 7 d @ 1 d + /// ~0.8 s vs 5.2–6.1 s, 1 d @ 1 min ~0.16 s vs 0.31–0.40 s (docs/Historian.md). + /// + /// + /// ⚠️ The key seeks are separate reads of the table within one statement, so under READ COMMITTED a sample + /// deleted (HistoryUpdate DeleteRaw/DeleteAtTime) between the aggregate and its seek leaves that pick NULL while + /// the counts still include it. The reader maps a NULL pick to "absent" (ReadStoredOrNull, live-gate 8 F1) + /// and the evaluator answers that bucket's Start/End/Delta BadNoData — transient, until the next read. A + /// sample replaced in between is picked with its new value (same time) — except that a usable pick re-checks + /// usability in its join (the exact u predicate), so a Replace that turned it Bad/Empty reads as absent + /// (BadNoData), never as a Bad row's value under Delta. A sample inserted in between is in neither half + /// (the pick times come from the aggregate). + /// + /// + /// Every join is also bounded to the window (the same @start/@end predicate as s): keyed + /// on the pick time alone, the optimizer may choose a hash/merge/adaptive join whose inner input is the tag's + /// whole history, and parameter sniffing then reuses that plan for every tag. + /// /// /// Schema-qualified Sample table name. /// The SQL text. /// True for a reverse read (window (@start, @end]). public static string StartEnd(string sample, bool reverse = false) { - static string Pick(string when, bool any) => - (any ? $"MAX(CASE WHEN {when} THEN SourceTimeUtc END), " : "") + - $"MAX(CASE WHEN {when} THEN ValueKind END), MAX(CASE WHEN {when} THEN NumValue END), " + - $"MAX(CASE WHEN {when} THEN IntValue END), MAX(CASE WHEN {when} THEN StrValue END), " + - (any ? $"MAX(CASE WHEN {when} THEN StatusCode END)" : $"MAX(CASE WHEN {when} THEN SourceTimeUtc END)"); + // A pick's columns from its joined row, in the established order: usable picks (kind, num, int, str, time), any- + // quality picks (time, kind, num, int, str, status). + static string Usable(string a) => $"{a}.ValueKind, {a}.NumValue, {a}.IntValue, {a}.StrValue, {a}.SourceTimeUtc"; + static string Any(string a) => $"{a}.SourceTimeUtc, {a}.ValueKind, {a}.NumValue, {a}.IntValue, {a}.StrValue, {a}.StatusCode"; + // Each seek is bounded to the window as well as keyed on the pick time: without the bounds the optimizer may pick a + // hash/merge/adaptive join whose inner input is the tag's WHOLE history (TagId = @id only), and parameter sniffing + // then reuses that plan for every tag (review of 910befc8: 43,657 logical reads vs 361 on a tag with 3M rows of + // history outside a 1 d window). A usable pick also re-checks usability (the exact `u` predicate), so a sample a + // HistoryUpdate Replace turned Bad/Empty in between joins nothing ⇒ absent ⇒ BadNoData, never a Bad row's value. + string Window(string a) => reverse + ? $"{a}.SourceTimeUtc > @start AND {a}.SourceTimeUtc <= @end" + : $"{a}.SourceTimeUtc >= @start AND {a}.SourceTimeUtc < @end"; + string Join(string a, string time, bool usable) => + $"LEFT JOIN {sample} {a} ON {a}.TagId = @id AND {a}.SourceTimeUtc = st.{time} AND {Window(a)}" + + (usable ? $" AND {a}.StatusCode >= 0 AND {a}.ValueKind <> 0" : "") + " "; return Classified(sample, reverse) + - ", w AS (SELECT b, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, u, g, " + - "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc ASC) AS rfa, " + - "ROW_NUMBER() OVER (PARTITION BY b ORDER BY SourceTimeUtc DESC) AS rla, " + - "ROW_NUMBER() OVER (PARTITION BY b, u ORDER BY SourceTimeUtc ASC) AS rfu, " + - "ROW_NUMBER() OVER (PARTITION BY b, u ORDER BY SourceTimeUtc DESC) AS rlu FROM s) " + - "SELECT b, " + CommonCounts() + ", " + - Pick("u = 1 AND rfu = 1", any: false) + ", " + - Pick("u = 1 AND rlu = 1", any: false) + ", " + - Pick("rfa = 1", any: true) + ", " + - Pick("rla = 1", any: true) + " " + - "FROM w GROUP BY b ORDER BY b;"; + ", st AS (SELECT b, " + CommonCounts() + ", " + + "MIN(CASE WHEN u = 1 THEN SourceTimeUtc END) AS fu, MAX(CASE WHEN u = 1 THEN SourceTimeUtc END) AS lu, " + + "MIN(SourceTimeUtc) AS fa, MAX(SourceTimeUtc) AS la FROM s GROUP BY b) " + + "SELECT st.b, st.TotalCount, st.UsableCount, st.GoodCount, st.BadCount, st.NonNumericCount, " + + Usable("x") + ", " + Usable("y") + ", " + Any("p") + ", " + Any("q") + " FROM st " + + Join("x", "fu", usable: true) + Join("y", "lu", usable: true) + Join("p", "fa", usable: false) + Join("q", "la", usable: false) + "ORDER BY st.b;"; } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index c47f8b20..4c7f1a44 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -462,10 +462,13 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource AggregateFamily.StartEnd => SqlAggregateQueries.StartEnd(sample, plan.Reversed), _ => SqlAggregateQueries.ActualTime(sample, aggregate == HistoryAggregateType.MaximumActualTime, plan.Reversed), }; - // Second result set: the tag's data range (earliest + latest sample) — the start/end-of-data Partial rules - // (#677/#680 — any per-sample aggregate but Average, either direction) and Count's in-range empty interval (#680). + // FIRST result set when needed: the tag's data range (earliest + latest sample) — the start/end-of-data Partial + // rules (#677/#680 — any per-sample aggregate but Average, either direction) and Count's in-range empty interval + // (#680). Read before the bucket statement (#683): under READ COMMITTED the two statements can see different + // rows, and a sample appended in between must reach the buckets rather than only the range — the opposite + // order turned such a bucket into a Good Count of 0. var needsRange = AggregateEvaluator.NeedsDataRange(aggregate, plan); - if (needsRange) sql += " " + SqlAggregateQueries.DataRange(sample); + if (needsRange) sql = SqlAggregateQueries.DataRange(sample) + " " + sql; await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false); await using var cmd = _conn.CreateCommand(conn, sql); cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id }); @@ -474,14 +477,18 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource cmd.Parameters.Add(new SqlParameter("@intervalTicks", SqlDbType.BigInt) { Value = plan.IntervalTicks }); cmd.Parameters.Add(new SqlParameter("@bucketAdjustTicks", SqlDbType.BigInt) { Value = plan.BucketIndexAdjustTicks }); await using var reader = await cmd.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false); + if (needsRange) + { + if (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) + { + if (!reader.IsDBNull(0)) earliest = SqlHistorianConnection.GetUtc(reader, 0); + if (!reader.IsDBNull(1)) latest = SqlHistorianConnection.GetUtc(reader, 1); + } + if (!await reader.NextResultAsync(cancellationToken).ConfigureAwait(false)) + throw new InvalidOperationException("The processed-read batch returned no bucket result set."); + } while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) rows.Add(ReadBucket(reader, family)); - if (needsRange && await reader.NextResultAsync(cancellationToken).ConfigureAwait(false) - && await reader.ReadAsync(cancellationToken).ConfigureAwait(false)) - { - if (!reader.IsDBNull(0)) earliest = SqlHistorianConnection.GetUtc(reader, 0); - if (!reader.IsDBNull(1)) latest = SqlHistorianConnection.GetUtc(reader, 1); - } } var result = AggregateEvaluator.EvaluateAll(aggregate, plan, rows, earliest, latest, now); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index 6c2b3afd..5c90931d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -62,6 +62,40 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend return new QuestDbConformanceBackend(fx, prefix, o, conn, boot); } + /// + /// A backend whose sample table has the pre-#683 shape — tag NOT indexed — as a deployment created before + /// the index existed has it. Its bootstrapper runs with AutoCreateSchema=false, so it leaves the table as is. + /// + /// The shared fixture. + /// The backend. + public static async Task CreatePre683Async(QuestDbFixture fx) + { + fx.SkipIfUnavailable(); + var prefix = fx.UniquePrefix(); + var o = fx.Options(prefix, autoCreate: false); + var conn = new QuestDbConnection(o); + await using (var c = await conn.OpenAsync(CancellationToken.None)) + { + foreach (var ddl in QuestDbTables.CreateAll(prefix, null)) + { + await using var cmd = conn.CreateCommand(c, ddl.Replace("tag SYMBOL INDEX", "tag SYMBOL", StringComparison.Ordinal)); + await cmd.ExecuteNonQueryAsync(); + } + } + var boot = new QuestDbSchemaBootstrapper(conn, o); + await boot.EnsureAsync(CancellationToken.None); + return new QuestDbConformanceBackend(fx, prefix, o, conn, boot); + } + + /// Whether this instance's sample table has its tag column indexed (#683). + /// The table_columns() flag. + public async Task TagIndexedAsync() + { + await using var c = await Conn.OpenAsync(CancellationToken.None); + await using var cmd = Conn.CreateCommand(c, QuestDbTables.TagIndexedQuery(Conn.SampleTable)); + return Convert.ToBoolean(await cmd.ExecuteScalarAsync()); + } + /// public long ResolutionTicks => 10; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs new file mode 100644 index 00000000..bc5ac9da --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs @@ -0,0 +1,197 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Conformance; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests; + +/// +/// #683: the sample table's tag symbol index — created with new tables, added by the bootstrapper to a table +/// from before it existed — and the data-range lookups it serves (the start/end-of-data Partial bits and Count's +/// in-range empty interval), which must answer exactly as they did from the unindexed scan: tombstones excluded, an +/// outage marker counted as a sample. +/// +[Collection(QuestDbCollection.Name)] +public sealed class TagIndexTests(QuestDbFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private const uint GoodCalc = 0x00000401u; + private const uint NoDataCalc = 0x809B0401u; + + private static DateTime At(double s) => ConformanceData.At(s); + + /// + /// Tag R: an outage marker at +0 (its earliest sample), 1 at +100, 2 at +200 (its latest live sample), a row + /// at +300 replaced by a tombstone and a tombstone alone at +400 — neither may extend the data range. + /// + private static async Task SeedAsync(QuestDbConformanceBackend b) + { + await b.SeedAsync("R", + (At(0), HistorianValue.Empty, ConformanceData.BadNoCommunication), + (At(100), HistorianValue.FromDouble(1), ConformanceData.Good), + (At(200), HistorianValue.FromDouble(2), ConformanceData.Good), + (At(300), HistorianValue.FromDouble(3), ConformanceData.Good)); + await b.InsertRowsAsync("R", + [ + (At(300), HistorianValue.Empty, HistorianStatusCodes.BadNoData, true), + (At(400), HistorianValue.Empty, HistorianStatusCodes.BadNoData, true), + ]); + } + + private static readonly HistoryAggregateType[] Aggregates = + [ + HistoryAggregateType.Count, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum, HistoryAggregateType.Average, + HistoryAggregateType.Range, HistoryAggregateType.Start, HistoryAggregateType.End, HistoryAggregateType.Delta, + HistoryAggregateType.MinimumActualTime, HistoryAggregateType.MaximumActualTime, + ]; + + private static readonly (double From, double To)[] Windows = [(-100, 100), (50, 450), (150, 250), (250, 500)]; + + /// Every per-sample aggregate over every window, forward and reverse, as comparable tuples. + private static async Task> ReadMatrixAsync(QuestDbConformanceBackend b) + { + var src = b.CreateDataSource(); + var all = new List(); + foreach (var a in Aggregates) + foreach (var (from, to) in Windows) + foreach (var reverse in new[] { false, true }) + { + var r = reverse + ? await src.ReadProcessedAsync("R", At(to), At(from), TimeSpan.FromSeconds(50), a, Ct) + : await src.ReadProcessedAsync("R", At(from), At(to), TimeSpan.FromSeconds(50), a, Ct); + all.AddRange(r.Samples.Select(s => + $"{a} [{from},{to}) rev={reverse} {s.SourceTimestampUtc:O} {s.Value} 0x{s.StatusCode:X8}")); + } + return all; + } + + private static async Task AwaitIndexedAsync(QuestDbConformanceBackend b) + { + var deadline = DateTime.UtcNow.AddSeconds(15); + while (!await b.TagIndexedAsync()) + { + if (DateTime.UtcNow > deadline) (await b.TagIndexedAsync()).ShouldBeTrue("the ADD INDEX was never applied"); + await Task.Delay(50); // the ALTER is a WAL transaction: visible once applied + } + } + + [Fact] + public async Task A_new_sample_table_is_created_with_the_tag_index() + { + await using var b = await QuestDbConformanceBackend.CreateConcreteAsync(fx); + (await b.TagIndexedAsync()).ShouldBeTrue(); + } + + [Fact] + public async Task A_refused_ADD_INDEX_does_not_fail_the_bootstrap() + { + fx.SkipIfUnavailable(); + // A sample table whose tag column cannot be indexed (VARCHAR, not SYMBOL): the ALTER is refused up front, which + // stands in for any refusal (e.g. no DDL rights). The index is optional, so the bootstrap must still succeed. + var p = fx.UniquePrefix(); + var o = fx.Options(p); + var conn = new QuestDbConnection(o); + await using (var c = await conn.OpenAsync(Ct)) + { + var ddl = QuestDbTables.CreateSample(p, null).Replace("tag SYMBOL INDEX", "tag VARCHAR", StringComparison.Ordinal); + await using var cmd = conn.CreateCommand(c, ddl); + await cmd.ExecuteNonQueryAsync(Ct); + await using var alter = conn.CreateCommand(c, QuestDbTables.AddTagIndex(conn.SampleTable)); + await Should.ThrowAsync(() => alter.ExecuteNonQueryAsync(Ct)); // precondition: really refused + } + var log = new ListLogger(); + await new QuestDbSchemaBootstrapper(conn, o, log).EnsureAsync(Ct); + log.Warnings.ShouldContain(m => m.Contains("Could not add the tag index", StringComparison.Ordinal) + && m.Contains(QuestDbTables.AddTagIndex(conn.SampleTable), StringComparison.Ordinal)); + } + + /// Captures the rendered Warning messages. + private sealed class ListLogger : Microsoft.Extensions.Logging.ILogger + { + private readonly System.Collections.Concurrent.ConcurrentQueue _warnings = new(); + public IReadOnlyCollection Warnings => _warnings; + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true; + public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, + TState state, Exception? exception, Func formatter) + { + if (logLevel == Microsoft.Extensions.Logging.LogLevel.Warning) _warnings.Enqueue(formatter(state, exception)); + } + } + + [Fact] + public async Task The_data_range_seeks_the_tag_index() + { + await using var b = await QuestDbConformanceBackend.CreateConcreteAsync(fx); + foreach (var sql in new[] + { + QuestDbAggregateQueries.EarliestSample(b.Conn.SampleTable), + QuestDbAggregateQueries.LatestSample(b.Conn.SampleTable), + }) + { + await using var c = await b.Conn.OpenAsync(Ct); + await using var cmd = b.Conn.CreateCommand(c, "EXPLAIN " + sql); + QuestDbConnection.BindText(cmd, "R"); + QuestDbConnection.BindTimestamp(cmd, At(0)); + var plan = new List(); + await using var r = await cmd.ExecuteReaderAsync(Ct); + while (await r.ReadAsync(Ct)) plan.Add(r.GetString(0)); + string.Join("\n", plan).ShouldContain("Index", Case.Sensitive, "a bound-parameter tag seek must use the symbol index"); + } + } + + [Fact] + public async Task The_data_range_excludes_tombstones_and_counts_outage_markers() + { + await using var b = await QuestDbConformanceBackend.CreateConcreteAsync(fx); + await SeedAsync(b); + var src = b.CreateDataSource(); + + // [+50, +450) by 50 s: data range [+0 outage marker, +200]; the +300/+400 tombstones must not extend it. + var count = (await src.ReadProcessedAsync("R", At(50), At(450), TimeSpan.FromSeconds(50), HistoryAggregateType.Count, Ct)).Samples; + count.Select(s => s.Value is null ? (long?)null : Convert.ToInt64(s.Value)).ShouldBe([0L, 1L, 0L, 1L, null, null, null, null]); + count.Select(s => s.StatusCode).ShouldBe( + [ + GoodCalc, GoodCalc, GoodCalc, + GoodCalc | HistorianInfoBits.HistorianPartial, // the data ends inside [+200, +250) + NoDataCalc, NoDataCalc, NoDataCalc, NoDataCalc, + ]); + + // [-100, +100): the outage marker at +0 starts the data range, so [-50, 0] reaches it (edges inclusive) ⇒ 0, Good. + var early = (await src.ReadProcessedAsync("R", At(-100), At(100), TimeSpan.FromSeconds(50), HistoryAggregateType.Count, Ct)).Samples; + early[0].StatusCode.ShouldBe(NoDataCalc); + Convert.ToInt64(early[1].Value).ShouldBe(0L); + early[1].StatusCode.ShouldBe(GoodCalc); + Convert.ToInt64(early[3].Value).ShouldBe(0L); + early[3].StatusCode.ShouldBe(GoodCalc); + } + + [Fact] + public async Task A_pre_683_table_gains_the_index_on_boot_and_every_processed_read_answers_the_same() + { + await using var b = await QuestDbConformanceBackend.CreatePre683Async(fx); + (await b.TagIndexedAsync()).ShouldBeFalse(); + await SeedAsync(b); + var unindexed = await ReadMatrixAsync(b); + + // AutoCreateSchema=false never alters (it only logs the statement to run). + await new QuestDbSchemaBootstrapper(b.Conn, b.Options).EnsureAsync(Ct); + await Task.Delay(300); + (await b.TagIndexedAsync()).ShouldBeFalse(); + + // AutoCreateSchema=true migrates; a second, concurrent-style re-send on the same column is harmless. + var migrate = fx.Options(b.Options.TablePrefix); + await new QuestDbSchemaBootstrapper(new QuestDbConnection(migrate), migrate).EnsureAsync(Ct); + await new QuestDbSchemaBootstrapper(new QuestDbConnection(migrate), migrate).EnsureAsync(Ct); + await AwaitIndexedAsync(b); + await b.Wal.WaitAsync(b.Conn.SampleTable, Ct); + + var indexed = await ReadMatrixAsync(b); + indexed.ShouldBe(unindexed); + + // Still writable and readable after the migration (the table was not suspended by the double ALTER). + await b.SeedAsync("R", (At(500), HistorianValue.FromDouble(5), ConformanceData.Good)); + (await b.ReadBackAsync("R")).Count.ShouldBe(4); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTablesTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTablesTests.cs index 748469d8..a8b5328b 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTablesTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTablesTests.cs @@ -8,17 +8,26 @@ public sealed class QuestDbTablesTests [Fact] public void Sample_ddl_without_ttl_is_pinned() => QuestDbTables.CreateSample("hist_", null).ShouldBe( - "CREATE TABLE IF NOT EXISTS hist_sample (ts TIMESTAMP, tag SYMBOL, kind BYTE, num_value DOUBLE, " + + "CREATE TABLE IF NOT EXISTS hist_sample (ts TIMESTAMP, tag SYMBOL INDEX, kind BYTE, num_value DOUBLE, " + "int_value LONG, str_value VARCHAR, status LONG, deleted BOOLEAN, server_ts TIMESTAMP) " + "TIMESTAMP(ts) PARTITION BY DAY WAL DEDUP UPSERT KEYS(ts, tag)"); [Fact] public void Sample_ddl_with_ttl_puts_ttl_between_partition_and_wal() => QuestDbTables.CreateSample("hist_", 7).ShouldBe( - "CREATE TABLE IF NOT EXISTS hist_sample (ts TIMESTAMP, tag SYMBOL, kind BYTE, num_value DOUBLE, " + + "CREATE TABLE IF NOT EXISTS hist_sample (ts TIMESTAMP, tag SYMBOL INDEX, kind BYTE, num_value DOUBLE, " + "int_value LONG, str_value VARCHAR, status LONG, deleted BOOLEAN, server_ts TIMESTAMP) " + "TIMESTAMP(ts) PARTITION BY DAY TTL 7 DAYS WAL DEDUP UPSERT KEYS(ts, tag)"); + [Fact] + public void Tag_index_migration_statements_are_pinned() + { + // #683: the migration for a sample table created before the tag index existed, and its probe. + QuestDbTables.AddTagIndex("hist_sample").ShouldBe("ALTER TABLE hist_sample ALTER COLUMN tag ADD INDEX"); + QuestDbTables.TagIndexedQuery("hist_sample").ShouldBe( + "SELECT indexed FROM table_columns('hist_sample') WHERE \"column\" = 'tag'"); + } + [Fact] public void Event_ddl_is_pinned() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/StartEndJoinBackTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/StartEndJoinBackTests.cs new file mode 100644 index 00000000..04e40336 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests/StartEndJoinBackTests.cs @@ -0,0 +1,123 @@ +using System.Data; +using Microsoft.Data.SqlClient; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests; + +/// +/// #683: aggregates the pick times, then fetches each pick by a key join. +/// The joins must be bounded to the window: keyed on the pick time alone, a hash/merge/adaptive join reads the tag's +/// whole history (review of 910befc8: 43,657 logical reads vs 361 on a tag with 3M rows outside a 1 d window). The +/// plan choice depends on estimates, so the test FORCES the worst shape (OPTION (HASH JOIN)) on a tag with long +/// history either side of the window and asserts the statement reads only window-sized pages — and returns exactly +/// what the unhinted plan and the data source return. +/// +[Collection(SqlHistorianCollection.Name)] +public sealed class StartEndJoinBackTests(SqlHistorianFixture fx) +{ + private static readonly CancellationToken Ct = CancellationToken.None; + private static DateTime At(double s) => ReadTestDb.At(s); + private const int HistoryRowsEachSide = 150_000; + + /// + /// Tag H: one sample a minute over the hour [+0, +3600) of mixed quality (Good, Uncertain, Bad, an outage + /// marker, a string), plus rows a second before +0 and after +3600. + /// + private static async Task<(ReadTestDb Db, int Id, List<(DateTime T, HistorianValue V, uint S)> Window)> SeedAsync(SqlHistorianFixture fx) + { + var db = await ReadTestDb.CreateAsync(fx); + var window = new List<(DateTime T, HistorianValue V, uint S)>(); + for (var i = 0; i < 60; i++) + { + var (v, s) = (i % 7) switch + { + 3 => (HistorianValue.FromDouble(i), ReadTestDb.Bad), + 4 => (HistorianValue.FromDouble(i), ReadTestDb.Uncertain), + 5 => (HistorianValue.Empty, ReadTestDb.BadNoCommunication), + 6 => (HistorianValue.FromString("s" + i), ReadTestDb.Good), + _ => (HistorianValue.FromDouble(i), ReadTestDb.Good), + }; + window.Add((At(i * 60 + 7), v, s)); + } + var id = await db.SeedAsync("H", window.ToArray()); + // The history: inserted set-based (a per-row INSERT of 300k rows would dominate the test). + await db.ScalarAsync( + $"WITH n AS (SELECT TOP ({HistoryRowsEachSide}) ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS x " + + "FROM sys.all_objects a CROSS JOIN sys.all_objects b) " + + $"INSERT {db.Conn.Q("Sample")} (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " + + $"SELECT {id}, DATEADD(second, -x, CAST('{At(0):yyyy-MM-dd HH:mm:ss.fffffff}' AS datetime2(7))), 3, x, NULL, NULL, 0, SYSUTCDATETIME() FROM n " + + $"UNION ALL SELECT {id}, DATEADD(second, x, CAST('{At(3600):yyyy-MM-dd HH:mm:ss.fffffff}' AS datetime2(7))), 3, x, NULL, NULL, 0, SYSUTCDATETIME() FROM n; " + + "SELECT 0"); + await db.ScalarAsync($"UPDATE STATISTICS {db.Conn.Q("Sample")}; SELECT 0"); + return (db, id, window); + } + + /// + /// Runs the StartEnd SQL (optionally with a query hint) and returns its rows as strings plus the session's logical + /// reads during the statement (sys.dm_exec_sessions.logical_reads before/after — STATISTICS IO messages do + /// not reach SqlClient's InfoMessage here). The two probes add a few reads of their own; the bound is far apart. + /// + private static async Task<(List Rows, long LogicalReads)> RunAsync(ReadTestDb db, int id, BucketPlan plan, string? hint) + { + var sql = SqlAggregateQueries.StartEnd(db.Conn.Q("Sample"), plan.Reversed); + if (hint is not null) sql = sql.TrimEnd().TrimEnd(';') + $" OPTION ({hint});"; + await using var c = await db.Conn.OpenAsync(Ct); + async Task SessionReadsAsync() + { + await using var probe = db.Conn.CreateCommand(c, "SELECT logical_reads FROM sys.dm_exec_sessions WHERE session_id = @@SPID"); + return Convert.ToInt64(await probe.ExecuteScalarAsync(Ct)); + } + var before = await SessionReadsAsync(); + await using var cmd = db.Conn.CreateCommand(c, sql); + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id }); + SqlHistorianConnection.AddDateTime2(cmd, "@start", plan.StartUtc); + SqlHistorianConnection.AddDateTime2(cmd, "@end", plan.EndUtc); + cmd.Parameters.Add(new SqlParameter("@intervalTicks", SqlDbType.BigInt) { Value = plan.IntervalTicks }); + cmd.Parameters.Add(new SqlParameter("@bucketAdjustTicks", SqlDbType.BigInt) { Value = plan.BucketIndexAdjustTicks }); + var rows = new List(); + await using (var r = await cmd.ExecuteReaderAsync(Ct)) + while (await r.ReadAsync(Ct)) + rows.Add(string.Join("|", Enumerable.Range(0, r.FieldCount).Select(i => r.IsDBNull(i) ? "NULL" : Convert.ToString(r.GetValue(i), System.Globalization.CultureInfo.InvariantCulture)))); + return (rows, await SessionReadsAsync() - before); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task The_pick_joins_read_only_the_window_even_under_a_forced_hash_join(bool reverse) + { + fx.SkipIfUnavailable(); + var (db, id, window) = await SeedAsync(fx); + var plan = reverse + ? BucketPlan.Create(At(3600), At(0), TimeSpan.FromMinutes(10)) + : BucketPlan.Create(At(0), At(3600), TimeSpan.FromMinutes(10)); + + var (plain, _) = await RunAsync(db, id, plan, hint: null); + var (hashed, hashReads) = await RunAsync(db, id, plan, hint: "HASH JOIN"); + var (merged, mergeReads) = await RunAsync(db, id, plan, hint: "MERGE JOIN"); + + plain.Count.ShouldBe(6); + hashed.ShouldBe(plain); + merged.ShouldBe(plain); + // Measured: 183 session logical reads bounded vs 8,055 with the window bound removed from the joins (300k history + // rows either side; the threshold sits ~5x above the one and ~8x below the other). Every join + // bounded ⇒ each input is a window-range seek, whatever the join algorithm. + hashReads.ShouldBeGreaterThan(0L, "the session read counter must move (positive control)"); + hashReads.ShouldBeLessThan(1000, "a hash join's inner input must be bounded to the window"); + mergeReads.ShouldBeLessThan(1000, "a merge join's inner input must be bounded to the window"); + + // And the picks are right: Start / End are each bucket's first / last sample of any quality. + var src = db.Source(); + var start = await src.ReadProcessedAsync("H", reverse ? At(3600) : At(0), reverse ? At(0) : At(3600), + TimeSpan.FromMinutes(10), HistoryAggregateType.Start, Ct); + var end = await src.ReadProcessedAsync("H", reverse ? At(3600) : At(0), reverse ? At(0) : At(3600), + TimeSpan.FromMinutes(10), HistoryAggregateType.End, Ct); + var buckets = Enumerable.Range(0, 6).Select(b => window.Skip(b * 10).Take(10).ToList()).ToList(); + if (reverse) buckets.Reverse(); + start.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(buckets.Select(b => (DateTime?)b[0].T)); + end.Samples.Select(s => s.SourceTimestampUtc).ShouldBe(buckets.Select(b => (DateTime?)b[^1].T)); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs index 24f40663..626db115 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests/BucketRowMappingTests.cs @@ -76,6 +76,25 @@ public sealed class BucketRowMappingTests AggregateEvaluator.Evaluate(agg, b, T, T).StatusCode.ShouldBe(NoDataCalc); } + [Fact] + public void A_pick_nulled_by_a_delete_racing_the_key_join_is_no_data_for_that_end_only() + { + // #683 join-back: the bucket's first sample (usable, so both first picks) was deleted between the GROUP BY and its + // key seek — the counts still include it, its pick columns are NULL; the last picks joined normally. + var t = StartEndTable(); + var row = Counts(t, 0, 2); + row["k1"] = (byte)1; row["n1"] = 20d; row["t1"] = T.AddSeconds(1); + row["at1"] = T.AddSeconds(1); row["ak1"] = (byte)1; row["an1"] = 20d; row["st1"] = 0; + t.Rows.Add(row); + + var b = Map(t, AggregateFamily.StartEnd); + b.FirstAny.ShouldBeNull(); + b.First.Kind.ShouldBe(HistorianValueKind.Empty); + AggregateEvaluator.Evaluate(HistoryAggregateType.Start, b, T, T).StatusCode.ShouldBe(NoDataCalc); + AggregateEvaluator.Evaluate(HistoryAggregateType.Delta, b, T, T).StatusCode.ShouldBe(NoDataCalc); + AggregateEvaluator.Evaluate(HistoryAggregateType.End, b, T, T).Value.ShouldBe(20d); + } + [Fact] public void A_consistent_row_still_maps_every_pick() {