Merge perf/683-data-range: QuestDB tag index, data range read first, SqlServer StartEnd window-bounded PK join-back (#683)
This commit is contained in:
Joseph Doherty
@@ -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.
|
||||
|
||||
+78
-8
@@ -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
|
||||
|
||||
+17
-11
@@ -450,7 +450,9 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource
|
||||
/// the SDK: window <c>(end, start]</c>, buckets stepping back from <c>start</c>, the partial one earliest, each stamped
|
||||
/// with its later edge, returned latest first. One statement per node returns the <see cref="BucketRow"/> statistics
|
||||
/// of each non-empty bucket (bucket index <c>(cast(ts AS LONG)*10 − BucketBaseUnixTicks) / IntervalTicks</c>, never
|
||||
/// <c>SAMPLE BY</c>); a second fetches the tag's earliest sample for the start-of-data Partial bit; and
|
||||
/// <c>SAMPLE BY</c>); before it, when <see cref="AggregateEvaluator.NeedsDataRange"/>, 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
|
||||
/// <see cref="AggregateEvaluator.EvaluateAll"/> produces the values and fills the empty buckets with <c>BadNoData</c>. The time-weighted family (#671) instead streams the window plus four bounding
|
||||
/// rows into <see cref="TimeWeightedAggregator"/>, exactly as the SQL Server backend does.
|
||||
/// </remarks>
|
||||
@@ -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);
|
||||
|
||||
+49
-1
@@ -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;
|
||||
/// <item><b>TTL (RC3):</b> <c>tables()</c> normalises units (7 days reads back as <c>1 WEEK</c>) 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.</item>
|
||||
/// <item><b>Tag index (#683):</b> a sample table created before the <c>tag</c> symbol index existed gains it with
|
||||
/// <see cref="QuestDbTables.AddTagIndex"/> (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.</item>
|
||||
/// <item><b>Failure:</b> not latched. Within <see cref="BootstrapFailureCache.DefaultDuration"/> the last failure is
|
||||
/// re-thrown without touching QuestDB.</item>
|
||||
/// <item><c>AutoCreateSchema=false</c> only verifies the tables exist (no TTL changes) and throws
|
||||
/// <see cref="HistorianSchemaMissingException"/> otherwise.</item>
|
||||
/// <see cref="HistorianSchemaMissingException"/> otherwise; a missing tag index is only logged (a Warning naming
|
||||
/// the ALTER to run), since reads stay correct without it.</item>
|
||||
/// </list>
|
||||
/// </summary>
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #683: indexes the sample table's <c>tag</c> column when it is not (a table created before the index existed).
|
||||
/// <paramref name="apply"/> false (AutoCreateSchema off) only logs the statement to run.
|
||||
/// </summary>
|
||||
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;
|
||||
|
||||
@@ -39,9 +39,15 @@ internal static class QuestDbTables
|
||||
/// </summary>
|
||||
public static string[] Expiring(string prefix) => [Sample(prefix), Event(prefix), SampleModified(prefix)];
|
||||
|
||||
/// <summary><c>CREATE TABLE IF NOT EXISTS</c> for the sample table.</summary>
|
||||
/// <summary>
|
||||
/// <c>CREATE TABLE IF NOT EXISTS</c> for the sample table. <c>tag</c> carries a symbol (bitmap) index (#683): every
|
||||
/// read filters on one tag, and without the index QuestDB answers a <c>tag = x</c> seek by scanning the tag column of
|
||||
/// every partition in the time range — the data-range lookups (<see cref="QuestDbAggregateQueries.EarliestSample"/> /
|
||||
/// <c>LatestSample</c>) 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 <see cref="AddTagIndex"/>.
|
||||
/// </summary>
|
||||
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)];
|
||||
|
||||
/// <summary>
|
||||
/// <c>ALTER TABLE t ALTER COLUMN tag ADD INDEX</c> — 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.
|
||||
/// </summary>
|
||||
public static string AddTagIndex(string table) => $"ALTER TABLE {table} ALTER COLUMN tag ADD INDEX";
|
||||
|
||||
/// <summary>
|
||||
/// Whether <paramref name="table"/>'s <c>tag</c> column is indexed: one row, one BOOLEAN column, no row when the
|
||||
/// table or column is missing. The name is inlined (<c>table_columns()</c> takes a constant); it is always a
|
||||
/// prefix-validated name built here.
|
||||
/// </summary>
|
||||
public static string TagIndexedQuery(string table) =>
|
||||
$"SELECT indexed FROM table_columns('{table}') WHERE \"column\" = 'tag'";
|
||||
|
||||
/// <summary><c>ALTER TABLE t SET TTL n DAYS</c>; <c>null</c> removes the TTL (<c>SET TTL 0 DAYS</c>).</summary>
|
||||
public static string AlterTtl(string table, int? retentionDays) =>
|
||||
$"ALTER TABLE {table} SET TTL {(retentionDays ?? 0).ToString(CultureInfo.InvariantCulture)} DAYS";
|
||||
|
||||
@@ -78,36 +78,59 @@ internal static class SqlAggregateQueries
|
||||
/// StatusCode</c>) and <c>22..27</c> the last (same order) — present for every returned (non-empty) bucket.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <b>One scan (live-gate 8 F1).</b> The counts and all four picked samples come from one reference to <c>s</c>:
|
||||
/// a windowed pass ranks each row within its bucket (<c>rfa</c>/<c>rla</c> over every sample, <c>rfu</c>/<c>rlu</c>
|
||||
/// within its usable / not-usable partition) and the outer <c>GROUP BY</c> both counts and lifts each picked row's
|
||||
/// columns with a conditional <c>MAX</c> (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.
|
||||
/// <para>
|
||||
/// <b>Aggregate, then join back on the key (#683).</b> One <c>GROUP BY</c> over the window computes the counts and
|
||||
/// each bucket's four pick TIMES — <c>MIN</c>/<c>MAX(SourceTimeUtc)</c> 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
|
||||
/// <c>(TagId, SourceTimeUtc)</c> (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 <c>ROW_NUMBER</c> pass + conditional
|
||||
/// <c>MAX</c>, live-gate 8 F1) — <c>EXCEPT</c> 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).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// ⚠️ 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" (<c>ReadStoredOrNull</c>, live-gate 8 F1)
|
||||
/// and the evaluator answers that bucket's Start/End/Delta <c>BadNoData</c> — 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 <c>u</c> predicate), so a Replace that turned it Bad/Empty reads as absent
|
||||
/// (<c>BadNoData</c>), 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).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Every join is also bounded to the window (the same <c>@start</c>/<c>@end</c> predicate as <c>s</c>): 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.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="sample">Schema-qualified <c>Sample</c> table name.</param>
|
||||
/// <returns>The SQL text.</returns>
|
||||
/// <param name="reverse">True for a reverse read (window <c>(@start, @end]</c>).</param>
|
||||
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;";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+16
-9
@@ -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);
|
||||
|
||||
+34
@@ -62,6 +62,40 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend
|
||||
return new QuestDbConformanceBackend(fx, prefix, o, conn, boot);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A backend whose sample table has the pre-#683 shape — <c>tag</c> NOT indexed — as a deployment created before
|
||||
/// the index existed has it. Its bootstrapper runs with <c>AutoCreateSchema=false</c>, so it leaves the table as is.
|
||||
/// </summary>
|
||||
/// <param name="fx">The shared fixture.</param>
|
||||
/// <returns>The backend.</returns>
|
||||
public static async Task<QuestDbConformanceBackend> 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);
|
||||
}
|
||||
|
||||
/// <summary>Whether this instance's sample table has its <c>tag</c> column indexed (#683).</summary>
|
||||
/// <returns>The <c>table_columns()</c> flag.</returns>
|
||||
public async Task<bool> 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());
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
public long ResolutionTicks => 10;
|
||||
|
||||
|
||||
+197
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// #683: the sample table's <c>tag</c> 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.
|
||||
/// </summary>
|
||||
[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);
|
||||
|
||||
/// <summary>
|
||||
/// Tag <c>R</c>: 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.
|
||||
/// </summary>
|
||||
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)];
|
||||
|
||||
/// <summary>Every per-sample aggregate over every window, forward and reverse, as comparable tuples.</summary>
|
||||
private static async Task<List<string>> ReadMatrixAsync(QuestDbConformanceBackend b)
|
||||
{
|
||||
var src = b.CreateDataSource();
|
||||
var all = new List<string>();
|
||||
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<Exception>(() => 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));
|
||||
}
|
||||
|
||||
/// <summary>Captures the rendered Warning messages.</summary>
|
||||
private sealed class ListLogger : Microsoft.Extensions.Logging.ILogger
|
||||
{
|
||||
private readonly System.Collections.Concurrent.ConcurrentQueue<string> _warnings = new();
|
||||
public IReadOnlyCollection<string> Warnings => _warnings;
|
||||
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
|
||||
public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
|
||||
public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId,
|
||||
TState state, Exception? exception, Func<TState, Exception?, string> 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<string>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
+11
-2
@@ -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()
|
||||
{
|
||||
|
||||
+123
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// #683: <see cref="SqlAggregateQueries.StartEnd"/> 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 (<c>OPTION (HASH JOIN)</c>) 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.
|
||||
/// </summary>
|
||||
[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;
|
||||
|
||||
/// <summary>
|
||||
/// Tag <c>H</c>: one sample a minute over the hour [+0, +3600) of mixed quality (Good, Uncertain, Bad, an outage
|
||||
/// marker, a string), plus <see cref="HistoryRowsEachSide"/> rows a second before +0 and after +3600.
|
||||
/// </summary>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs the StartEnd SQL (optionally with a query hint) and returns its rows as strings plus the session's logical
|
||||
/// reads during the statement (<c>sys.dm_exec_sessions.logical_reads</c> 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.
|
||||
/// </summary>
|
||||
private static async Task<(List<string> 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<long> 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<string>();
|
||||
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));
|
||||
}
|
||||
}
|
||||
+19
@@ -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()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user