diff --git a/docs/Historian.md b/docs/Historian.md index 77368e11f..14cceeb48 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -1940,9 +1940,9 @@ 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`; **`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}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, 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`) (#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 @@ -1962,10 +1962,13 @@ slower on long windows**: the index scan is single-threaded where the unindexed 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. +**`{p}event.source` and `{p}sample_modified.tag` are indexed too** (they were deliberately left unindexed at #683, on the grounds +that both reads are window-bounded and both tables are small). A HistoryReadEvents of one notifier filters `source = x` and every +`IsReadModified` read filters `tag = x`; unindexed, each scans that column in every partition of the window — on a long window of a busy +alarm table that is the same multi-partition scan #683 measured costing seconds on the sample table. Both are created with the index now, and +the bootstrapper adds a missing one exactly as it does the sample `tag` index (`ALTER TABLE {p}event ALTER COLUMN source ADD INDEX`, +`ALTER TABLE {p}sample_modified ALTER COLUMN tag ADD INDEX`; same idempotence, same Warning-only posture with `AutoCreateSchema=false` or a +refused ALTER). The trade-off is the one measured above (disk + O3 rewrite cost), on tables far smaller than the sample table. `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, 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 8564e729e..8286b6635 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 @@ -16,8 +16,9 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; /// TTL (RC3): tables() normalises units (7 days reads back as 1 WEEK) so values are compared /// in hours, and an ALTER becomes visible only after the WAL applies it, so the bootstrapper never re-reads to verify /// its own ALTER. A boot whose TTL already matches issues no ALTER. -/// Tag index: a sample table created before the tag symbol index existed gains it with -/// (idempotent; a boot that finds it indexed issues no ALTER, and like the TTL +/// Symbol indexes: a table created before its symbol index existed (the sample and modification-trail +/// tag, the event source — ) gains it with +/// (idempotent; a boot that finds it indexed issues no ALTER, and like the TTL /// the ALTER is a WAL transaction the bootstrapper does not re-read). Building it is a one-off WAL apply — ~3 s for /// 68M rows / 63 daily partitions measured — during which writes queue and reads keep scanning. A refused ALTER is /// logged as a Warning, never fails the bootstrap (the index is optional); a failure at WAL-apply time would suspend @@ -71,12 +72,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); + await EnsureIndexesAsync(conn, apply: true, ct).ConfigureAwait(false); } else { await VerifyAsync(conn, ct).ConfigureAwait(false); - await EnsureTagIndexAsync(conn, apply: false, ct).ConfigureAwait(false); + await EnsureIndexesAsync(conn, apply: false, ct).ConfigureAwait(false); } _ready = true; _failure.Clear(); @@ -127,23 +128,30 @@ internal sealed class QuestDbSchemaBootstrapper } } - /// - /// Indexes the sample table's tag column when it is not (a table created before the index existed). - /// false (AutoCreateSchema off) only logs the statement to run. - /// - private async Task EnsureTagIndexAsync(NpgsqlConnection conn, bool apply, CancellationToken ct) + /// Ensures every column is indexed (see ). + private async Task EnsureIndexesAsync(NpgsqlConnection conn, bool apply, CancellationToken ct) { - var table = _conn.SampleTable; - await using (var cmd = _conn.CreateCommand(conn, QuestDbTables.TagIndexedQuery(table))) + foreach (var (table, column) in QuestDbTables.SymbolIndexes(_conn.TablePrefix)) + await EnsureIndexAsync(conn, table, column, apply, ct).ConfigureAwait(false); + } + + /// + /// Indexes 's symbol when it is not (a table created before the + /// index existed). false (AutoCreateSchema off) only logs the statement to run. + /// + private async Task EnsureIndexAsync(NpgsqlConnection conn, string table, string column, bool apply, CancellationToken ct) + { + var statement = QuestDbTables.AddIndex(table, column); + await using (var cmd = _conn.CreateCommand(conn, QuestDbTables.IndexedQuery(table, column))) { 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)); + _log.LogWarning("QuestDB historian table {Table} has no index on its {Column} column, so reads filtering on it " + + "scan partitions instead of seeking. AutoCreateSchema is false; run: {Statement}", + table, column, statement); return; } // A WAL transaction: table_columns() keeps reporting false until it applies, so do NOT re-read to verify. @@ -152,17 +160,17 @@ internal sealed class QuestDbSchemaBootstrapper // suspend the table) — the same blind spot as the TTL ALTER. try { - await using var alter = _conn.CreateCommand(conn, QuestDbTables.AddTagIndex(table)); + await using var alter = _conn.CreateCommand(conn, statement); 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)); + _log.LogWarning("Could not add the {Column} index to QuestDB table {Table} ({Exception}: {Message}); reads " + + "filtering on it scan partitions instead of seeking. Run: {Statement}", + column, table, ex.GetType().Name, ex.Message, statement); return; } - _log.LogInformation("Added the tag index to QuestDB table {Table} (#683).", table); + _log.LogInformation("Added the {Column} index to QuestDB table {Table}.", column, table); } private async Task ReconcileTtlAsync(NpgsqlConnection conn, 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 4fe1f025b..5fd00eaa1 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 @@ -66,12 +66,16 @@ internal static class QuestDbTables "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)"; - /// CREATE TABLE IF NOT EXISTS for the event table. + /// + /// CREATE TABLE IF NOT EXISTS for the event table. source carries a symbol index: a HistoryReadEvents + /// of one notifier filters on it, and unindexed that filter scans every partition in the window. A table created + /// before the index existed gains it through . + /// /// The configured table-name prefix. /// Retention in days, or null for no TTL. /// The DDL statement. public static string CreateEvent(string prefix, int? retentionDays) => - $"CREATE TABLE IF NOT EXISTS {Event(prefix)} (ts TIMESTAMP, event_id VARCHAR, source SYMBOL, alarm_id VARCHAR, " + + $"CREATE TABLE IF NOT EXISTS {Event(prefix)} (ts TIMESTAMP, event_id VARCHAR, source SYMBOL INDEX, alarm_id VARCHAR, " + "alarm_name VARCHAR, alarm_type_name VARCHAR, severity INT, event_kind VARCHAR, message VARCHAR, " + "user_name VARCHAR, comment VARCHAR, received_utc TIMESTAMP) " + $"TIMESTAMP(ts) PARTITION BY DAY{Ttl(retentionDays)} WAL DEDUP UPSERT KEYS(ts, event_id)"; @@ -87,13 +91,14 @@ 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. + /// 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). /// /// The configured table-name prefix. /// Retention in days, or null for no TTL. /// The DDL statement. public static string CreateSampleModified(string prefix, int? retentionDays) => - $"CREATE TABLE IF NOT EXISTS {SampleModified(prefix)} (ts TIMESTAMP, tag SYMBOL, modified_id LONG, " + + $"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"; @@ -106,6 +111,29 @@ internal static class QuestDbTables [CreateSample(prefix, retentionDays), CreateEvent(prefix, retentionDays), CreateTag(prefix), CreateSampleModified(prefix, retentionDays)]; + /// + /// The symbol columns that carry an index, as (table, column): the sample table's tag (#683), the event + /// table's source and the modification trail's tag. The bootstrapper adds any that a table created before + /// the index existed lacks. + /// + /// The configured table-name prefix. + /// The indexed columns. + public static (string Table, string Column)[] SymbolIndexes(string prefix) => + [(Sample(prefix), "tag"), (Event(prefix), "source"), (SampleModified(prefix), "tag")]; + + /// ALTER TABLE t ALTER COLUMN c ADD INDEX (idempotent on an indexed column, a WAL transaction). + /// The table name. + /// The symbol column. + /// The ALTER statement. + public static string AddIndex(string table, string column) => $"ALTER TABLE {table} ALTER COLUMN {column} ADD INDEX"; + + /// Whether 's is indexed (one BOOLEAN row; none when missing). + /// The table name (prefix-validated, inlined — table_columns() takes a constant). + /// The column name (a constant from ). + /// The query text. + public static string IndexedQuery(string table, string column) => + $"SELECT indexed FROM table_columns('{table}') WHERE \"column\" = '{column}'"; + /// /// ALTER TABLE t ALTER COLUMN tag ADD INDEX — the 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 diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs index 7c8857f0d..62b9e1999 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs @@ -78,7 +78,8 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend { foreach (var ddl in QuestDbTables.CreateAll(prefix, null)) { - await using var cmd = conn.CreateCommand(c, ddl.Replace("tag SYMBOL INDEX", "tag SYMBOL", StringComparison.Ordinal)); + // Tables from before ANY symbol index existed: the sample / trail tag and the event source unindexed. + await using var cmd = conn.CreateCommand(c, ddl.Replace(" SYMBOL INDEX", " SYMBOL", StringComparison.Ordinal)); await cmd.ExecuteNonQueryAsync(); } } @@ -87,6 +88,17 @@ internal sealed class QuestDbConformanceBackend : IConformanceBackend return new QuestDbConformanceBackend(fx, prefix, o, conn, boot); } + /// Whether 's is indexed. + /// The table. + /// The symbol column. + /// The table_columns() flag. + public async Task IndexedAsync(string table, string column) + { + await using var c = await Conn.OpenAsync(CancellationToken.None); + await using var cmd = Conn.CreateCommand(c, QuestDbTables.IndexedQuery(table, column)); + return Convert.ToBoolean(await cmd.ExecuteScalarAsync()); + } + /// Whether this instance's sample table has its tag column indexed (#683). /// The table_columns() flag. public async Task TagIndexedAsync() diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs index bc5ac9dae..2b7e05eaf 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs @@ -83,6 +83,42 @@ public sealed class TagIndexTests(QuestDbFixture fx) (await b.TagIndexedAsync()).ShouldBeTrue(); } + [Fact] + public async Task New_tables_index_every_filtered_symbol_column() + { + await using var b = await QuestDbConformanceBackend.CreateConcreteAsync(fx); + foreach (var (table, column) in QuestDbTables.SymbolIndexes(b.Options.TablePrefix)) + (await b.IndexedAsync(table, column)).ShouldBeTrue($"{table}.{column}"); + } + + [Fact] + public async Task Pre_index_event_and_trail_tables_gain_their_indexes_on_boot_and_event_reads_answer_the_same() + { + await using var b = await QuestDbConformanceBackend.CreatePre683Async(fx); + var p = b.Options.TablePrefix; + foreach (var (table, column) in QuestDbTables.SymbolIndexes(p)) + (await b.IndexedAsync(table, column)).ShouldBeFalse($"precondition: {table}.{column} unindexed"); + await b.SeedEventsAsync((At(10), "e1", "S1"), (At(20), "e2", "S2"), (At(30), "e3", "S1")); + var src = b.CreateDataSource(); + async Task ReadAsync() => string.Join(",", + (await src.ReadEventsAsync("S1", At(0), At(100), 0, Ct)).Events.Select(e => $"{e.EventId}@{e.EventTimeUtc:O}")); + var before = await ReadAsync(); + before.ShouldNotBeEmpty(); + + var migrate = fx.Options(p); + await new QuestDbSchemaBootstrapper(new QuestDbConnection(migrate), migrate).EnsureAsync(Ct); + var deadline = DateTime.UtcNow.AddSeconds(15); + foreach (var (table, column) in QuestDbTables.SymbolIndexes(p)) + { + while (!await b.IndexedAsync(table, column)) + { + DateTime.UtcNow.ShouldBeLessThan(deadline, $"{table}.{column} was never indexed"); + await Task.Delay(50); + } + } + (await ReadAsync()).ShouldBe(before); + } + [Fact] public async Task A_refused_ADD_INDEX_does_not_fail_the_bootstrap() { 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 a8b5328ba..41af62e69 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 @@ -28,11 +28,26 @@ public sealed class QuestDbTablesTests "SELECT indexed FROM table_columns('hist_sample') WHERE \"column\" = 'tag'"); } + [Fact] + public void Every_filtered_symbol_column_is_indexed_and_has_a_migration() + { + QuestDbTables.SymbolIndexes("hist_").ShouldBe( + [("hist_sample", "tag"), ("hist_event", "source"), ("hist_sample_modified", "tag")]); + QuestDbTables.AddIndex("hist_event", "source").ShouldBe("ALTER TABLE hist_event ALTER COLUMN source ADD INDEX"); + QuestDbTables.IndexedQuery("hist_event", "source").ShouldBe( + "SELECT indexed FROM table_columns('hist_event') WHERE \"column\" = 'source'"); + // The CREATE statements carry every index the migration would add. + var ddl = QuestDbTables.CreateAll("hist_", null); + foreach (var (table, column) in QuestDbTables.SymbolIndexes("hist_")) + ddl.Single(d => d.StartsWith($"CREATE TABLE IF NOT EXISTS {table} (", StringComparison.Ordinal)) + .ShouldContain($"{column} SYMBOL INDEX"); + } + [Fact] public void Event_ddl_is_pinned() { QuestDbTables.CreateEvent("hist_", null).ShouldBe( - "CREATE TABLE IF NOT EXISTS hist_event (ts TIMESTAMP, event_id VARCHAR, source SYMBOL, alarm_id VARCHAR, " + + "CREATE TABLE IF NOT EXISTS hist_event (ts TIMESTAMP, event_id VARCHAR, source SYMBOL INDEX, alarm_id VARCHAR, " + "alarm_name VARCHAR, alarm_type_name VARCHAR, severity INT, event_kind VARCHAR, message VARCHAR, " + "user_name VARCHAR, comment VARCHAR, received_utc TIMESTAMP) " + "TIMESTAMP(ts) PARTITION BY DAY WAL DEDUP UPSERT KEYS(ts, event_id)"); @@ -57,7 +72,7 @@ public sealed class QuestDbTablesTests public void Sample_modified_ddl_is_pinned_append_only_and_carries_the_ttl() { QuestDbTables.CreateSampleModified("hist_", null).ShouldBe( - "CREATE TABLE IF NOT EXISTS hist_sample_modified (ts TIMESTAMP, tag SYMBOL, modified_id LONG, " + + "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");