diff --git a/CLAUDE.md b/CLAUDE.md index fc86bc04..dd2ba426 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. `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..da348f9a 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,22 @@ 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. 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), and the per-sample / raw / time-weighted window queries were unchanged or +slightly faster (the tag filter becomes an index scan). + `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 +1581,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 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..a775f7f8 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,15 @@ 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. /// 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 +69,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 +125,31 @@ 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. + await using (var alter = _conn.CreateCommand(conn, QuestDbTables.AddTagIndex(table))) + await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + _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/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..25b19b36 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs @@ -0,0 +1,160 @@ +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 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() {