diff --git a/docs/Historian.md b/docs/Historian.md index fad1348c9..92df06cb6 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -79,7 +79,7 @@ consumes the historian. The existing enable gates (`ServerHistorian:Enabled`, `A | Comms-loss / driver-Bad publishes | Dropped, counted in `DroppedUnsupportedKind` | Stored as an `Empty` outage-marker sample carrying the Bad code (see [below](#typed-values-real-status-and-outage-markers)) | Stored as an `Empty` outage-marker sample, as on `SqlServer` | | Processed aggregates | Twelve: the ten per-sample ones (Average, Count, Minimum, Maximum, Range, Start, End, Delta, Min/MaxActualTime) evaluated from raw reads with the SQL backends' evaluator — identical results (#681) — plus TimeAverage and Total from AVEVA's native modes | The seventeen in the [aggregate table](#aggregates-processed-reads) (incl. Part 13 time-weighted, #671) | The same seventeen, identical results | | HistoryUpdate | No (`BadHistoryOperationUnsupported`) | Yes (Insert / Replace / Update / DeleteRaw / DeleteAtTime) | Yes — soft delete (tombstones), see [below](#historyupdate-soft-delete) | -| Modified-value history (`IsReadModified`, #668) | No (`BadHistoryOperationUnsupported`) | Yes — `SampleModified`, atomic with the change | Yes — `{prefix}sample_modified`, written before the change (orphan-on-crash) | +| Modified-value history (`IsReadModified`, #668) | No (`BadHistoryOperationUnsupported`) | Yes — `SampleModified`, atomic with the change | Yes — `{prefix}sample_modified`, written before the change, marked applied after it (an orphan is not served) | | Alarm history | gateway `SendEvent`, one RPC per event, up to 8 in flight (no batch RPC yet) | `Event` table (insert-if-absent), multi-row statements of 150 events | `{prefix}event` table (dedup upsert on `(ts, event_id)`), multi-row INSERTs of 500 events | | Provisioning (`EnsureTags`) | numeric types only; others skipped | `MERGE` upsert of every request, all types | Upsert into `{prefix}tag`, all types | | Connection | `Endpoint` + `ApiKey` + TLS keys (top-level `ServerHistorian:*`) | `ServerHistorian:SqlServer:*` | `ServerHistorian:QuestDb:*` | @@ -1094,9 +1094,12 @@ modification time and the Part 11 `HistoryUpdateType`. A HistoryReadRaw with `Is - **Atomicity differs by backend.** `SqlServer` writes the trail row with `OUTPUT … INTO` in the **same statement** that changes `hist.Sample`, so a change and its trail row commit or roll back together. QuestDB has no multi-table transaction, so the trail row is written **before** the change: a crash (or failed sample INSERT) in between leaves an **orphan trail row** describing a change that never landed — never a - change without its trail row. Treat a QuestDB trail entry whose sample still shows the prior value as such an orphan. - A failed sample write after its trail rows landed logs a **Warning** naming the tag, the row count and the time range, so the orphan - can be found. + change without its trail row. A failed sample write after its trail rows landed logs a **Warning** naming the tag, the row count and the + time range. **Applied marker:** the trail row is written `pending = true` and, once the change has landed, re-upserted `pending = false` on + the table's `DEDUP UPSERT KEYS(ts, tag, modified_id)`. A modified read serves a still-`pending` row only when it is the **latest** trail row at + its time and the sample there (a tombstone included) was written at or after the modification — a crash between the change and its marker. + Otherwise it is the orphan of a change that never landed and is **not served** (it used to be served as a change that happened). Residual: + a double fault — a change that landed without its marker, followed by a later change of the same sample — hides the earlier one. - **Rolling upgrade (SqlServer).** The v2 schema is applied by the first node that upgrades. Until its peer is upgraded too, that not-yet-upgraded node still runs the old build, whose HistoryUpdates write **no trail rows** — silent gaps in the audit trail. Upgrade both nodes of a pair together (stop, upgrade, start both); modifications made by an old-build node in the window are never recorded. @@ -1116,11 +1119,15 @@ modification time and the Part 11 `HistoryUpdateType`. A HistoryReadRaw with `Is decided at materialise time from "an updater is wired", and whether the schema has the trail is only known after the backend's lazy first-use bootstrap (I/O the node manager does not do while materialising), so a v1 node is admitted to HistoryUpdate and then refused. Apply `002_sample_modified.sql` and restart the node to enable them (the verdict is taken once per process). `AutoCreateSchema=true` (the default) still upgrades v1 to v2 in place. See [Schema](#schema). -- **QuestDB retry duplicates (#674).** `{p}sample_modified` is append-only with **no DEDUP key** (a key would collapse genuinely distinct - modifications of one sample), and each trail row gets a fresh `modified_id`. So a HistoryUpdate a client **retries** after a timeout — when the - first attempt had in fact landed — records a second trail row for the same change (and, for Insert, the retry is `BadEntryExists`). The - sample itself is idempotent (dedup upsert); only the audit trail can show the duplicate. `SqlServer` has the same exposure only if the client - retries a committed request; there is no request id to dedupe on in either backend. +- **QuestDB retry duplicates (#674).** `{p}sample_modified` carries **`DEDUP UPSERT KEYS(ts, tag, modified_id)`** (each modification keeps its + own `modified_id`, so distinct modifications of one sample never collapse). The updater re-sends a trail INSERT once, **with the same ids**, + after a transient failure (a client timeout that may have applied): it converges instead of duplicating. A HistoryUpdate the **client** + retries is a new request with new ids, so it still records a second trail row when the first attempt had in fact landed (and, for Insert, + the retry is `BadEntryExists`). `SqlServer` has the same client-retry exposure; there is no request id to dedupe on in either backend. +- **QuestDB trail migration.** A trail table from before the marker gains `pending` (old rows read `false`, so they stay visible) and the + dedup key on first boot with `AutoCreateSchema=true` (`ALTER TABLE … ADD COLUMN pending BOOLEAN`, `ALTER TABLE … DEDUP ENABLE UPSERT KEYS(ts, + tag, modified_id)`). Unlike the index ALTERs this one is awaited (≤ 10 s), because an INSERT naming the new column would fail before it + applies; until then, and with `AutoCreateSchema=false` (a Warning naming the statements), the old format is used — no marker, no retry. - **Cross-node ordering (QuestDB).** The per-tag lock is per process; the [Primary-only](#primary-only-674) HistoryUpdate gate (#674) is what keeps two nodes of a pair from recording concurrent updates of one tag (and so a wrong prior value). Two **separate** pairs sharing one QuestDB can still race; that is unsupported. @@ -1942,7 +1949,7 @@ makes HistoryUpdate Replace work; the writer's own insert-if-absent semantics ar | `{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`; **`source` is a `SYMBOL INDEX`**) | `(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` (**`SYMBOL INDEX`**)`, 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) | +| `{p}sample_modified` (`ts` = the sample's source time, `tag` (**`SYMBOL INDEX`**)`, modified_id, modification_ts, update_type, principal, api_key_id, kind, num_value, int_value, str_value, status, server_ts, pending`) (#668) | `(ts, tag, modified_id)` — each modification is its own row; the key only lets the applied marker and a re-sent INSERT converge. `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 @@ -2045,7 +2052,8 @@ present **and not deleted**. **Modification trail (#668).** Each operation first appends one `{p}sample_modified` row per sample it will change, carrying the live row's current value (read by the same existence query), then writes the change. No multi-table transaction exists, so this order is the guarantee: a crash between the two leaves an **orphan trail row**, never an untracked change — the documented deviation from `SqlServer`'s atomic trail. -See [Modified-value history](#modified-value-history-part-11). +The row is written `pending` and marked applied after the change, so a modified read does not serve the orphan of a change that never +landed. See [Modified-value history](#modified-value-history-part-11). The per-item codes come from the shared `HistoryUpdatePlanner`, so they equal `SqlServer`'s (including the duplicate-timestamp-in-one-request rules above). **Every read and aggregate filters `NOT deleted`.** A client Insert at a deleted timestamp re-populates it (the OPC UA intent); only the 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 49b814d60..12c98f992 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 @@ -192,25 +192,44 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource var start = QuestDbConnection.Round(startUtc); var end = QuestDbConnection.Round(endUtc); var reverse = start > end; - string where; - if (start == end) where = "ts = $2 AND ts = $3"; - else if (!reverse) where = "ts >= $2 AND ts < $3"; - else where = "ts > $2 AND ts <= $3"; + string window; + if (start == end) window = "ts = $2 AND ts = $3"; + else if (!reverse) window = "ts >= $2 AND ts < $3"; + else window = "ts > $2 AND ts <= $3"; + var resume = ""; var next = 4; if (resumeAfter is not null) { - where += reverse + resume = reverse ? " AND (ts < $4 OR (ts = $4 AND modified_id < $5))" : " AND (ts > $4 OR (ts = $4 AND modified_id > $5))"; next = 6; } var order = reverse ? "DESC" : "ASC"; var (lo, hi) = start <= end ? (start, end) : (end, start); + const string Columns = "ts, kind, num_value, int_value, str_value, status, server_ts, modified_id, " + + "modification_ts, update_type, principal"; + + // The applied marker: a trail row still `pending` is served only when the change it describes evidently + // landed — it is the LATEST row at its time and the sample there (a tombstone included) was written at or + // after the modification: a crash between the change and its marker. Otherwise it is the orphan of a change + // that never landed (typically followed by the client's successful retry, which is then the latest row) and + // is dropped. The ASOF JOIN reads the tag's samples of the window only (indexed tag, inclusive bounds); the + // resume position filters OUTSIDE the window function so it never cuts a time's partition. + var sql = _bootstrap.SupportsTrailMarker + ? $"SELECT {Columns} FROM (SELECT m.ts ts, m.kind kind, m.num_value num_value, m.int_value int_value, " + + "m.str_value str_value, m.status status, m.server_ts server_ts, m.modified_id modified_id, " + + "m.modification_ts modification_ts, m.update_type update_type, m.principal principal, m.pending pending, " + + "s.ts sample_ts, s.server_ts sample_server_ts, max(m.modified_id) OVER (PARTITION BY m.ts) latest_id " + + $"FROM {_conn.SampleModifiedTable} m ASOF JOIN (SELECT ts, tag, server_ts FROM {_conn.SampleTable} " + + $"WHERE tag = $1 AND ts >= $2 AND ts <= $3) s ON (tag) WHERE m.tag = $1 AND {window.Replace("ts ", "m.ts ")}) " + + "WHERE (NOT pending OR (sample_ts = ts AND sample_server_ts >= modification_ts AND modified_id = latest_id))" + + $"{resume} ORDER BY ts {order}, modified_id {order}" + : $"SELECT {Columns} FROM {_conn.SampleModifiedTable} " + + $"WHERE tag = $1 AND {window}{resume} ORDER BY ts {order}, modified_id {order}"; await using var conn = await _conn.OpenAsync(cancellationToken).ConfigureAwait(false); - await using var cmd = _conn.CreateCommand(conn, - $"SELECT {SampleColumns}, modified_id, modification_ts, update_type, principal FROM {_conn.SampleModifiedTable} " + - $"WHERE tag = $1 AND {where} ORDER BY ts {order}, modified_id {order}" + (bounded ? $" LIMIT ${next}" : "")); + await using var cmd = _conn.CreateCommand(conn, sql + (bounded ? $" LIMIT ${next}" : "")); QuestDbConnection.BindText(cmd, fullReference); QuestDbConnection.BindTimestamp(cmd, lo); QuestDbConnection.BindTimestamp(cmd, hi); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs index bbe0def25..fd6e1ae81 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs @@ -32,6 +32,13 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// no multi-table transaction, so the trail row is written BEFORE the change: a crash (or a failed sample INSERT) /// between the two leaves an orphan trail row describing a change that never landed — never a change without its /// trail row. This is the documented deviation from SQL Server, whose trail commits atomically with the change. +/// Applied marker. On a trail table with the marker format +/// () the trail row is written pending and, once the +/// sample change has landed, re-upserted with pending = false (same (ts, tag, modified_id) — the table's +/// dedup key). A modified read drops a pending row unless the sample at its time was written at or after the +/// modification (the change landed but the marker did not — a crash in between), so an orphan is no longer served as +/// a change that happened. The trail INSERT is re-sent once (same ids) after a transient failure: a first attempt that +/// did apply converges through the dedup key instead of duplicating. /// internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater { @@ -154,10 +161,11 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater if (page.Count == 0) break; // Trail BEFORE the change (no multi-table transaction): a crash here leaves an orphan trail row, never a // tombstone without its trail row. - await WriteTrailAsync(tag, HistoryModificationType.Delete, + var trail = await WriteTrailAsync(tag, HistoryModificationType.Delete, priors.Select(r => (r.TimeUtc, (LiveRow?)r)).ToList(), context, ct).ConfigureAwait(false); await AfterTrailAsync(tag, page[0], page[^1], page.Count, () => WriteTombstonesAsync(tag, page, ct)).ConfigureAwait(false); + await MarkAppliedAsync(tag, trail, ct).ConfigureAwait(false); // The wait between pages is required: without it the next page could re-read rows just tombstoned. await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); total += page.Count; @@ -197,11 +205,12 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater foreach (var chunk in toDelete.Chunk(BatchSize)) { // Trail BEFORE the change (see the class remarks). - await WriteTrailAsync(tag, HistoryModificationType.Delete, + var trail = await WriteTrailAsync(tag, HistoryModificationType.Delete, chunk.Select(t => (live[t].TimeUtc, (LiveRow?)live[t])).ToList(), context, ct).ConfigureAwait(false); var chunkTimes = chunk.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(); await AfterTrailAsync(tag, chunkTimes[0], chunkTimes[^1], chunkTimes.Count, () => WriteTombstonesAsync(tag, chunkTimes, ct)).ConfigureAwait(false); + await MarkAppliedAsync(tag, trail, ct).ConfigureAwait(false); } if (toDelete.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); deleted.UnionWith(toDelete); @@ -233,10 +242,11 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater trail.Add((stored, present ? prior : null)); } // Trail BEFORE the change (see the class remarks). - await WriteTrailAsync(tag, ToModificationType(mode), trail, context, ct).ConfigureAwait(false); + var written = await WriteTrailAsync(tag, ToModificationType(mode), trail, context, ct).ConfigureAwait(false); foreach (var chunk in toWrite.Chunk(BatchSize)) await AfterTrailAsync(tag, chunk[0].TimeUtc, chunk[^1].TimeUtc, chunk.Length, () => WriteRowsAsync(tag, chunk, ct)).ConfigureAwait(false); + await MarkAppliedAsync(tag, written, ct).ConfigureAwait(false); if (toWrite.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false); return outcome; } @@ -268,7 +278,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater private const string LiveColumns = "ts, kind, num_value, int_value, str_value, status, server_ts"; /// A live sample as it was before a modification. - private readonly record struct LiveRow(DateTime TimeUtc, HistorianValue Value, uint Status, DateTime? ServerTimeUtc); + internal readonly record struct LiveRow(DateTime TimeUtc, HistorianValue Value, uint Status, DateTime? ServerTimeUtc); private static LiveRow ReadLiveRow(NpgsqlDataReader rd) { @@ -305,63 +315,38 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater } } + /// One modification-trail row as written: its key, its prior value and the fields re-sent by the marker. + internal sealed record TrailRow(DateTime TimeUtc, LiveRow? Prior, long ModifiedId, HistoryModificationType Type, + DateTime ModificationUtc, string Principal, string? ApiKeyId); + + /// Test seam: runs after a trail INSERT executed (once per attempt), before its outcome is decided. + internal Func? AfterTrailInsert { get; set; } + + /// Test seam: runs before each sample write that follows a written trail batch (a throw = a failed change). + internal Func? BeforeSampleWrite { get; set; } + + /// Test seam: when true the applied marker is never written (a crash after the change, before the marker). + internal bool SkipAppliedMarker { get; set; } + /// /// Appends one modification-trail row per entry (prior = null ⇒ an insert: kind Empty, NULL value and - /// status), then waits for the trail's WAL visibility so a modified read right after the update sees it. + /// status) — pending on a marker-format trail — then waits for the trail's WAL visibility so a modified read + /// right after the update sees it. Returns the rows written, for . /// - private async Task WriteTrailAsync(string tag, HistoryModificationType type, IReadOnlyList<(DateTime TimeUtc, LiveRow? Prior)> entries, - HistoryUpdateContext context, CancellationToken ct) + private async Task> WriteTrailAsync(string tag, HistoryModificationType type, + IReadOnlyList<(DateTime TimeUtc, LiveRow? Prior)> entries, HistoryUpdateContext context, CancellationToken ct) { - if (entries.Count == 0) return; + if (entries.Count == 0) return []; var now = DateTime.UtcNow; - foreach (var chunk in entries.Chunk(BatchSize)) + var rows = entries + .Select(e => new TrailRow(e.TimeUtc, e.Prior, NextModifiedId(), type, now, context.StoredPrincipal, context.StoredApiKeyId)) + .ToList(); + var marker = _tags.Bootstrap.SupportsTrailMarker; + foreach (var chunk in rows.Chunk(BatchSize)) { - await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); - await using var cmd = _conn.CreateCommand(conn, ""); - var sb = new System.Text.StringBuilder( - $"INSERT INTO {_conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, update_type, principal, " + - "api_key_id, kind, num_value, int_value, str_value, status, server_ts) VALUES "); - for (var i = 0; i < chunk.Length; i++) - { - var (time, prior) = chunk[i]; - QuestDbConnection.BindTimestamp(cmd, time, clampToEpoch: true); - QuestDbConnection.BindText(cmd, tag); - QuestDbConnection.BindLong(cmd, NextModifiedId()); - QuestDbConnection.BindTimestamp(cmd, now); - QuestDbConnection.BindByte(cmd, (byte)type); - QuestDbConnection.BindText(cmd, context.StoredPrincipal); - QuestDbConnection.BindText(cmd, context.StoredApiKeyId); - if (prior is { } p) - { - var c = SampleColumnMapper.ToColumns(p.Value); - QuestDbConnection.BindByte(cmd, c.ValueKind); - Typed(cmd, NpgsqlDbType.Double, c.NumValue); - Typed(cmd, NpgsqlDbType.Bigint, c.IntValue); - Typed(cmd, NpgsqlDbType.Varchar, c.StrValue); - QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(p.Status)); - Typed(cmd, NpgsqlDbType.TimestampTz, p.ServerTimeUtc); - } - else - { - // No prior value: QuestDB BYTE has no NULL, so kind is Empty (0) and the NULL status marks "none". - QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty); - Typed(cmd, NpgsqlDbType.Double, null); - Typed(cmd, NpgsqlDbType.Bigint, null); - Typed(cmd, NpgsqlDbType.Varchar, null); - Typed(cmd, NpgsqlDbType.Bigint, null); - Typed(cmd, NpgsqlDbType.TimestampTz, null); - } - - var b = i * 13; - if (i > 0) sb.Append(','); - sb.Append('('); - for (var q = 1; q <= 13; q++) sb.Append(q > 1 ? "," : "").Append('$').Append(b + q); - sb.Append(')'); - } - cmd.CommandText = sb.ToString(); try { - await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + await InsertTrailAsync(tag, chunk, pending: marker, retryOnce: marker, ct).ConfigureAwait(false); } catch (Exception ex) when (ex is not OperationCanceledException) { @@ -371,6 +356,104 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater } } await _wal.WaitAsync(_conn.SampleModifiedTable, ct).ConfigureAwait(false); + return rows; + } + + /// + /// The applied marker: re-upserts the trail rows of a change that has landed with pending = false (same + /// (ts, tag, modified_id) key). A failure here does not fail the update — the change landed, and a modified read + /// still serves a pending row whose sample shows the change (logged as a Warning). + /// + private async Task MarkAppliedAsync(string tag, IReadOnlyList rows, CancellationToken ct) + { + if (rows.Count == 0 || !_tags.Bootstrap.SupportsTrailMarker || SkipAppliedMarker) return; + try + { + foreach (var chunk in rows.Chunk(BatchSize)) + await InsertTrailAsync(tag, chunk, pending: false, retryOnce: true, ct).ConfigureAwait(false); + await _wal.WaitAsync(_conn.SampleModifiedTable, ct).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _log.LogWarning(ex, "QuestDB applied marker for {Count} trail row(s) of tag {Tag} was not written; the change landed and " + + "modified reads still serve it.", rows.Count, tag); + } + } + + private int _trailInserts; + + /// Trail INSERT attempts so far (lifetime) — test/diagnostic. + internal int TrailInserts => Volatile.Read(ref _trailInserts); + + /// One multi-row trail INSERT; re-sent once with the same values after a transient failure when allowed. + private async Task InsertTrailAsync(string tag, IReadOnlyList chunk, bool pending, bool retryOnce, CancellationToken ct) + { + for (var attempt = 0; ; attempt++) + { + try + { + await InsertTrailOnceAsync(tag, chunk, pending, ct).ConfigureAwait(false); + Interlocked.Increment(ref _trailInserts); + if (AfterTrailInsert is { } hook) await hook(attempt).ConfigureAwait(false); + return; + } + catch (Exception ex) when (retryOnce && attempt == 0 && !ct.IsCancellationRequested && QuestDbConnection.IsTransient(ex)) + { + // The first attempt may have applied (a client timeout after the server took it): the same ids re-sent + // converge through DEDUP UPSERT KEYS(ts, tag, modified_id) instead of duplicating. + _log.LogDebug("QuestDB modification-trail INSERT failed transiently ({Exception}); re-sending it once.", ex.GetType().Name); + } + } + } + + private async Task InsertTrailOnceAsync(string tag, IReadOnlyList chunk, bool pending, CancellationToken ct) + { + var columns = _tags.Bootstrap.SupportsTrailMarker ? 14 : 13; + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, ""); + var sb = new System.Text.StringBuilder( + $"INSERT INTO {_conn.SampleModifiedTable} (ts, tag, modified_id, modification_ts, update_type, principal, " + + "api_key_id, kind, num_value, int_value, str_value, status, server_ts" + (columns == 14 ? ", pending" : "") + ") VALUES "); + for (var i = 0; i < chunk.Count; i++) + { + var row = chunk[i]; + QuestDbConnection.BindTimestamp(cmd, row.TimeUtc, clampToEpoch: true); + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindLong(cmd, row.ModifiedId); + QuestDbConnection.BindTimestamp(cmd, row.ModificationUtc); + QuestDbConnection.BindByte(cmd, (byte)row.Type); + QuestDbConnection.BindText(cmd, row.Principal); + QuestDbConnection.BindText(cmd, row.ApiKeyId); + if (row.Prior is { } p) + { + var c = SampleColumnMapper.ToColumns(p.Value); + QuestDbConnection.BindByte(cmd, c.ValueKind); + Typed(cmd, NpgsqlDbType.Double, c.NumValue); + Typed(cmd, NpgsqlDbType.Bigint, c.IntValue); + Typed(cmd, NpgsqlDbType.Varchar, c.StrValue); + QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(p.Status)); + Typed(cmd, NpgsqlDbType.TimestampTz, p.ServerTimeUtc); + } + else + { + // No prior value: QuestDB BYTE has no NULL, so kind is Empty (0) and the NULL status marks "none". + QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty); + Typed(cmd, NpgsqlDbType.Double, null); + Typed(cmd, NpgsqlDbType.Bigint, null); + Typed(cmd, NpgsqlDbType.Varchar, null); + Typed(cmd, NpgsqlDbType.Bigint, null); + Typed(cmd, NpgsqlDbType.TimestampTz, null); + } + if (columns == 14) Typed(cmd, NpgsqlDbType.Boolean, pending); + + var b = i * columns; + if (i > 0) sb.Append(','); + sb.Append('('); + for (var q = 1; q <= columns; q++) sb.Append(q > 1 ? "," : "").Append('$').Append(b + q); + sb.Append(')'); + } + cmd.CommandText = sb.ToString(); + await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false); } /// @@ -382,6 +465,7 @@ internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater { try { + if (BeforeSampleWrite is { } hook) await hook().ConfigureAwait(false); await write().ConfigureAwait(false); } catch (Exception ex) 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 8286b6635..9d0dacaa0 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 @@ -73,11 +73,13 @@ internal sealed class QuestDbSchemaBootstrapper await CreateTablesAsync(conn, ct).ConfigureAwait(false); await ReconcileTtlAsync(conn, ct).ConfigureAwait(false); await EnsureIndexesAsync(conn, apply: true, ct).ConfigureAwait(false); + SupportsTrailMarker = await EnsureTrailMarkerAsync(conn, apply: true, ct).ConfigureAwait(false); } else { await VerifyAsync(conn, ct).ConfigureAwait(false); await EnsureIndexesAsync(conn, apply: false, ct).ConfigureAwait(false); + SupportsTrailMarker = await EnsureTrailMarkerAsync(conn, apply: false, ct).ConfigureAwait(false); } _ready = true; _failure.Clear(); @@ -128,6 +130,93 @@ internal sealed class QuestDbSchemaBootstrapper } } + private volatile bool _supportsTrailMarker; + + /// + /// Whether the modification trail has its pending column and DEDUP UPSERT KEYS(ts, tag, modified_id) + /// (true for every table created since; a pre-marker table once this boot's migration has applied). False ⇒ the + /// updater writes the trail the old way (no marker, no retry) and modified reads do not reconcile — set by + /// . + /// + internal bool SupportsTrailMarker + { + get => _supportsTrailMarker; + private set => _supportsTrailMarker = value; + } + + /// How long a boot waits for its own trail migration (two WAL ALTERs) to apply before using the old format. + internal static readonly TimeSpan TrailMigrationWait = TimeSpan.FromSeconds(10); + + /// + /// Gives a pre-marker trail table its pending column and dedup key (), then reports + /// whether both are in effect. The ALTERs are WAL transactions, and an INSERT naming a column the table does not have + /// yet would fail, so — unlike the TTL / index ALTERs — this one IS awaited (bounded by + /// ); until it applies the old format is used. false only + /// logs the statements to run. A refused ALTER is a Warning, never a bootstrap failure. + /// + private async Task EnsureTrailMarkerAsync(NpgsqlConnection conn, bool apply, CancellationToken ct) + { + var table = _conn.SampleModifiedTable; + var (hasColumn, hasDedup) = await TrailStateAsync(conn, table, ct).ConfigureAwait(false); + if (hasColumn && hasDedup) return true; + var statements = new List(); + if (!hasColumn) statements.Add(QuestDbTables.AddTrailPendingColumn(table)); + if (!hasDedup) statements.Add(QuestDbTables.EnableTrailDedup(table)); + if (!apply) + { + _log.LogWarning("QuestDB modification trail {Table} predates its applied marker, so modified reads cannot drop the " + + "orphan of a HistoryUpdate that failed after its trail row, and a retried trail INSERT can duplicate. " + + "AutoCreateSchema is false; run: {Statements}", table, string.Join("; ", statements)); + return false; + } + try + { + foreach (var sql in statements) + { + await using var alter = _conn.CreateCommand(conn, sql); + await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false); + } + } + catch (Exception ex) when (ex is not OperationCanceledException || !ct.IsCancellationRequested) + { + _log.LogWarning("Could not migrate QuestDB modification trail {Table} ({Exception}: {Message}); the trail keeps its " + + "old format. Run: {Statements}", table, ex.GetType().Name, ex.Message, string.Join("; ", statements)); + return false; + } + var deadline = DateTime.UtcNow + TrailMigrationWait; + while (true) + { + (hasColumn, hasDedup) = await TrailStateAsync(conn, table, ct).ConfigureAwait(false); + if (hasColumn && hasDedup) + { + _log.LogInformation("Migrated QuestDB modification trail {Table} to the applied-marker format.", table); + return true; + } + if (DateTime.UtcNow >= deadline) + { + _log.LogWarning("QuestDB modification trail {Table} migration did not apply within {Seconds} s; the old trail " + + "format is used until the next boot.", table, TrailMigrationWait.TotalSeconds); + return false; + } + await Task.Delay(50, ct).ConfigureAwait(false); + } + } + + private async Task<(bool Column, bool Dedup)> TrailStateAsync(NpgsqlConnection conn, string table, CancellationToken ct) + { + bool column; + await using (var cmd = _conn.CreateCommand(conn, QuestDbTables.ColumnCountQuery(table, "pending"))) + column = Convert.ToInt64(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false), CultureInfo.InvariantCulture) > 0; + bool dedup; + await using (var cmd = _conn.CreateCommand(conn, "SELECT dedup FROM tables() WHERE table_name = $1")) + { + QuestDbConnection.BindText(cmd, table); + var o = await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false); + dedup = o is not null and not DBNull && Convert.ToBoolean(o, CultureInfo.InvariantCulture); + } + return (column, dedup); + } + /// Ensures every column is indexed (see ). private async Task EnsureIndexesAsync(NpgsqlConnection conn, bool apply, CancellationToken ct) { 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 5fd00eaa1..b48397304 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 @@ -91,8 +91,13 @@ internal static class QuestDbTables /// CREATE TABLE IF NOT EXISTS for the Part 11 modification trail: one row per sample a HistoryUpdate /// inserted, replaced or deleted, carrying the PRIOR value (NULL kind/value/status for an insert). ts is the /// sample's source time (so the TTL ages it out with the samples); modified_id orders the trail within one - /// source time. Append-only: no DEDUP, never updated or deleted. tag carries a symbol index (every modified - /// read filters on one tag; see for the migration of an older table). + /// source time. Never updated or deleted; tag carries a symbol index (every modified read filters on one tag; + /// see for the migration of an older table). + /// DEDUP UPSERT KEYS(ts, tag, modified_id) + pending: a trail row is written pending = true + /// BEFORE its sample change and re-upserted pending = false (the applied marker) after it, so a modified read + /// can drop the orphan of a change that never landed; and a trail INSERT re-sent after a client timeout (same ids) + /// converges instead of duplicating. A table from before both gains them through + /// / . /// /// The configured table-name prefix. /// Retention in days, or null for no TTL. @@ -100,8 +105,25 @@ internal static class QuestDbTables public static string CreateSampleModified(string prefix, int? retentionDays) => $"CREATE TABLE IF NOT EXISTS {SampleModified(prefix)} (ts TIMESTAMP, tag SYMBOL INDEX, modified_id LONG, " + "modification_ts TIMESTAMP, update_type BYTE, principal VARCHAR, api_key_id VARCHAR, kind BYTE, num_value DOUBLE, " + - "int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP) " + - $"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL"; + "int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP, pending BOOLEAN) " + + $"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL DEDUP UPSERT KEYS(ts, tag, modified_id)"; + + /// The migration of a pre-marker trail table: ALTER TABLE t ADD COLUMN pending BOOLEAN (old rows read false). + /// The modification-trail table name. + /// The ALTER statement. + public static string AddTrailPendingColumn(string table) => $"ALTER TABLE {table} ADD COLUMN pending BOOLEAN"; + + /// The migration of a pre-marker trail table: ALTER TABLE t DEDUP ENABLE UPSERT KEYS(ts, tag, modified_id). + /// The modification-trail table name. + /// The ALTER statement. + public static string EnableTrailDedup(string table) => $"ALTER TABLE {table} DEDUP ENABLE UPSERT KEYS(ts, tag, modified_id)"; + + /// Whether has : one count() row. + /// The table name (prefix-validated, inlined). + /// The column (a constant). + /// The query text. + public static string ColumnCountQuery(string table, string column) => + $"SELECT count() FROM table_columns('{table}') WHERE \"column\" = '{column}'"; /// The four CREATE statements in creation order (parallel to ). /// The configured table-name prefix. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs index 8076eded6..650f5da10 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTagCatalog.cs @@ -33,6 +33,9 @@ internal sealed class QuestDbTagCatalog _clock = clock ?? TimeProvider.System; } + /// The schema bootstrapper this catalog runs (its ). + internal QuestDbSchemaBootstrapper Bootstrap => _bootstrap; + /// Read-only lookup; never creates a row. /// Historian tag name. /// Cancellation. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs new file mode 100644 index 000000000..745e159ba --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TrailMarkerTests.cs @@ -0,0 +1,154 @@ +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.QuestDb.IntegrationTests; + +/// +/// The QuestDB modification trail's applied marker and dedup key. QuestDB has no multi-table transaction, so the trail +/// row is written BEFORE its change: a failed change left an orphan row that modified reads served as a change that +/// happened, and a trail INSERT re-sent after a client timeout duplicated. Now the row is written pending, +/// marked applied after the change, keyed (ts, tag, modified_id); a modified read drops a pending row unless the +/// sample shows the change landed. +/// +[Collection(QuestDbCollection.Name)] +public sealed class TrailMarkerTests(QuestDbFixture fx) +{ + private static readonly DateTime T0 = new(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc); + private static readonly CancellationToken Ct = CancellationToken.None; + + private static QuestDbHistorianDataSource Source(UpdateTests.Rig rig) => new(rig.Conn, rig.Tags.Bootstrap, rig.Tags); + + private static async Task> ModifiedAsync(UpdateTests.Rig rig, string tag) => + (await Source(rig).ReadModifiedAsync(tag, T0.AddMinutes(-1), T0.AddMinutes(10), 0, null, Ct)).Values; + + private static async Task TrailRowsAsync(UpdateTests.Rig rig, string tag) + { + await using var conn = await rig.Conn.OpenAsync(Ct); + await using var cmd = rig.Conn.CreateCommand(conn, $"SELECT count() FROM {rig.Conn.SampleModifiedTable} WHERE tag = $1"); + QuestDbConnection.BindText(cmd, tag); + return Convert.ToInt64(await cmd.ExecuteScalarAsync()); + } + + [Fact] + public async Task A_new_trail_table_has_the_marker_format_and_an_applied_change_reads_back() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct)) + .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); + rig.Tags.Bootstrap.SupportsTrailMarker.ShouldBeTrue(); + + var trail = await ModifiedAsync(rig, "t"); + trail.ShouldHaveSingleItem().UpdateType.ShouldBe(HistoryModificationType.Insert); + (await TrailRowsAsync(rig, "t")).ShouldBe(1, "the applied marker re-upserts the row, it does not add one"); + } + + [Fact] + public async Task The_orphan_of_a_change_that_never_landed_is_not_served() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct); + + rig.Updater.BeforeSampleWrite = () => throw new InvalidOperationException("the sample INSERT failed"); + await Should.ThrowAsync( + () => rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct)); + (await TrailRowsAsync(rig, "t")).ShouldBe(2, "precondition: the orphan trail row is stored"); + + var trail = await ModifiedAsync(rig, "t"); + trail.ShouldHaveSingleItem("only the insert that happened").UpdateType.ShouldBe(HistoryModificationType.Insert); + + // Positive control: the same replace, landing, is served. + rig.Updater.BeforeSampleWrite = null; + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct); + (await ModifiedAsync(rig, "t")).Select(v => v.UpdateType) + .ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace]); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task A_change_that_landed_without_its_marker_is_still_served(bool delete) + { + // A crash between the sample change and the marker: the row stays pending, but the sample shows the change. + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct); + rig.Updater.SkipAppliedMarker = true; + + if (delete) + (await rig.Updater.DeleteAtTimeAsync("t", [T0], UpdateTests.Who, Ct)).ShouldBe([HistorianStatusCodes.Good]); + else + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct); + + (await ModifiedAsync(rig, "t")).Select(v => v.UpdateType).ShouldBe( + [HistoryModificationType.Insert, delete ? HistoryModificationType.Delete : HistoryModificationType.Replace]); + } + + [Fact] + public async Task A_trail_insert_re_sent_after_a_transient_failure_does_not_duplicate() + { + fx.SkipIfUnavailable(); + var rig = UpdateTests.MakeRig(fx); + await rig.Tags.Bootstrap.EnsureAsync(Ct); + var failedOnce = false; + // The first attempt APPLIES and then fails as a client timeout would: the re-send carries the same ids. + rig.Updater.AfterTrailInsert = attempt => + { + if (attempt == 0 && !failedOnce) { failedOnce = true; throw new TimeoutException("client timeout"); } + return Task.CompletedTask; + }; + + (await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, + [UpdateTests.V(T0, 1), UpdateTests.V(T0.AddSeconds(1), 2), UpdateTests.V(T0.AddSeconds(2), 3)], UpdateTests.Who, Ct)) + .ShouldAllBe(s => s == HistorianStatusCodes.GoodEntryInserted); + + failedOnce.ShouldBeTrue(); + rig.Updater.TrailInserts.ShouldBeGreaterThanOrEqualTo(3, "pending, the re-send, then the marker"); + (await TrailRowsAsync(rig, "t")).ShouldBe(3, "the re-sent rows converge on their dedup key"); + (await ModifiedAsync(rig, "t")).Count.ShouldBe(3); + } + + [Fact] + public async Task A_pre_marker_trail_table_is_migrated_on_boot_and_its_old_rows_stay_visible() + { + fx.SkipIfUnavailable(); + var p = fx.UniquePrefix(); + var o = fx.Options(p); + var c = new QuestDbConnection(o); + await using (var conn = await c.OpenAsync(Ct)) + { + foreach (var ddl in QuestDbTables.CreateAll(p, null)) + { + // The trail as it was: no pending column, no dedup key. + var old = ddl.Replace(", pending BOOLEAN", "", StringComparison.Ordinal) + .Replace(" DEDUP UPSERT KEYS(ts, tag, modified_id)", "", StringComparison.Ordinal); + await using var cmd = c.CreateCommand(conn, old); + await cmd.ExecuteNonQueryAsync(Ct); + } + } + + // An old-format update first (the bootstrap with AutoCreateSchema=false keeps the old format). + var legacyOptions = fx.Options(p, autoCreate: false); + var legacyConn = new QuestDbConnection(legacyOptions); + var legacyWal = new QuestDbWalVisibility(legacyConn, legacyOptions.VisibilityTimeout); + var legacyTags = new QuestDbTagCatalog(legacyConn, new QuestDbSchemaBootstrapper(legacyConn, legacyOptions), legacyWal); + var legacy = new UpdateTests.Rig(legacyConn, new QuestDbHistorianDataUpdater(legacyConn, legacyTags, legacyWal), legacyTags); + (await legacy.Updater.UpdateDataAsync("t", HistoryUpdateMode.Insert, [UpdateTests.V(T0, 1)], UpdateTests.Who, Ct)) + .ShouldBe([HistorianStatusCodes.GoodEntryInserted]); + legacyTags.Bootstrap.SupportsTrailMarker.ShouldBeFalse(); + (await ModifiedAsync(legacy, "t")).Count.ShouldBe(1); + + // A boot with AutoCreateSchema=true migrates; the old row (pending reads false) is still served, a new one too. + var wal = new QuestDbWalVisibility(c, o.VisibilityTimeout); + var tags = new QuestDbTagCatalog(c, new QuestDbSchemaBootstrapper(c, o), wal); + var rig = new UpdateTests.Rig(c, new QuestDbHistorianDataUpdater(c, tags, wal), tags); + await tags.Bootstrap.EnsureAsync(Ct); + tags.Bootstrap.SupportsTrailMarker.ShouldBeTrue(); + await rig.Updater.UpdateDataAsync("t", HistoryUpdateMode.Replace, [UpdateTests.V(T0, 2)], UpdateTests.Who, Ct); + (await ModifiedAsync(rig, "t")).Select(v => v.UpdateType) + .ShouldBe([HistoryModificationType.Insert, HistoryModificationType.Replace]); + } +} 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 41af62e69..05efd457f 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 @@ -69,14 +69,17 @@ public sealed class QuestDbTablesTests } [Fact] - public void Sample_modified_ddl_is_pinned_append_only_and_carries_the_ttl() + public void Sample_modified_ddl_is_pinned_with_its_applied_marker_and_dedup_key_and_carries_the_ttl() { QuestDbTables.CreateSampleModified("hist_", null).ShouldBe( "CREATE TABLE IF NOT EXISTS hist_sample_modified (ts TIMESTAMP, tag SYMBOL INDEX, modified_id LONG, " + "modification_ts TIMESTAMP, update_type BYTE, principal VARCHAR, api_key_id VARCHAR, kind BYTE, num_value DOUBLE, " + - "int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP) TIMESTAMP(ts) PARTITION BY DAY WAL"); - QuestDbTables.CreateSampleModified("hist_", 30).ShouldEndWith("PARTITION BY DAY TTL 30 DAYS WAL"); - QuestDbTables.CreateSampleModified("hist_", null).ShouldNotContain("DEDUP"); + "int_value LONG, str_value VARCHAR, status LONG, server_ts TIMESTAMP, pending BOOLEAN) TIMESTAMP(ts) PARTITION BY DAY " + + "WAL DEDUP UPSERT KEYS(ts, tag, modified_id)"); + QuestDbTables.CreateSampleModified("hist_", 30).ShouldEndWith("PARTITION BY DAY TTL 30 DAYS WAL DEDUP UPSERT KEYS(ts, tag, modified_id)"); + QuestDbTables.AddTrailPendingColumn("hist_sample_modified").ShouldBe("ALTER TABLE hist_sample_modified ADD COLUMN pending BOOLEAN"); + QuestDbTables.EnableTrailDedup("hist_sample_modified") + .ShouldBe("ALTER TABLE hist_sample_modified DEDUP ENABLE UPSERT KEYS(ts, tag, modified_id)"); } [Fact]