dohertj2/lmxopcua
1
0
Fork 0
Code Issues 31 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

perf(historian): index QuestDB event.source and sample_modified.tag (lmxopcua#746)

Browse Source 
Event reads filtered on source and every modified-value read filtered on tag
scanned that symbol column in every partition of the window; only
sample.tag carried an index (#683).

Both columns are now SYMBOL INDEX in the DDL, and the bootstrapper's index
migration is generalised over QuestDbTables.SymbolIndexes (sample.tag,
event.source, sample_modified.tag): a missing index is added with the same
idempotent ALTER ... ADD INDEX, Warning-only when AutoCreateSchema is false
or the ALTER is refused. Integration tests: new tables carry all three; a
pre-index table set gains them on boot and event reads answer the same.

This reverses #683's documented choice to leave the two unindexed;
docs/Historian.md carries the reasoning.

Rebased from the unpushed remed/lane-H2 branch (originally authored 2026-10-22
in session_01Ub9hGmYLQvn9XETzw7CMsT) onto current master.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-10-22 17:00:54 -04:00
parent d44e58dcd6
commit bc2f5e1889
6 changed files with 134 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/Historian.md
+9 -6
View File
@@ -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,
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbSchemaBootstrapper.cs
+27 -19
View File
@@ -16,8 +16,9 @@ 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:</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
/// <item><b>Symbol indexes:</b> a table created before its symbol index existed (the sample and modification-trail
/// <c>tag</c>, the event <c>source</c> — <see cref="QuestDbTables.SymbolIndexes"/>) gains it with
/// <see cref="QuestDbTables.AddIndex"/> (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
}
}
/// <summary>
/// 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)
/// <summary>Ensures every <see cref="QuestDbTables.SymbolIndexes"/> column is indexed (see <see cref="EnsureIndexAsync"/>).</summary>
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);
}
/// <summary>
/// Indexes <paramref name="table"/>'s symbol <paramref name="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 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)
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbTables.cs
+32 -4
View File
@@ -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)";
/// <summary><c>CREATE TABLE IF NOT EXISTS</c> for the event table.</summary>
/// <summary>
/// <c>CREATE TABLE IF NOT EXISTS</c> for the event table. <c>source</c> 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 <see cref="SymbolIndexes"/>.
/// </summary>
/// <param name="prefix">The configured table-name prefix.</param>
/// <param name="retentionDays">Retention in days, or null for no TTL.</param>
/// <returns>The DDL statement.</returns>
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
/// <c>CREATE TABLE IF NOT EXISTS</c> 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). <c>ts</c> is the
/// sample's source time (so the TTL ages it out with the samples); <c>modified_id</c> 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. <c>tag</c> carries a symbol index (every modified
/// read filters on one tag; see <see cref="SymbolIndexes"/> for the migration of an older table).
/// </summary>
/// <param name="prefix">The configured table-name prefix.</param>
/// <param name="retentionDays">Retention in days, or null for no TTL.</param>
/// <returns>The DDL statement.</returns>
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)];
/// <summary>
/// The symbol columns that carry an index, as <c>(table, column)</c>: the sample table's <c>tag</c> (#683), the event
/// table's <c>source</c> and the modification trail's <c>tag</c>. The bootstrapper adds any that a table created before
/// the index existed lacks.
/// </summary>
/// <param name="prefix">The configured table-name prefix.</param>
/// <returns>The indexed columns.</returns>
public static (string Table, string Column)[] SymbolIndexes(string prefix) =>
[(Sample(prefix), "tag"), (Event(prefix), "source"), (SampleModified(prefix), "tag")];
/// <summary><c>ALTER TABLE t ALTER COLUMN c ADD INDEX</c> (idempotent on an indexed column, a WAL transaction).</summary>
/// <param name="table">The table name.</param>
/// <param name="column">The symbol column.</param>
/// <returns>The ALTER statement.</returns>
public static string AddIndex(string table, string column) => $"ALTER TABLE {table} ALTER COLUMN {column} ADD INDEX";
/// <summary>Whether <paramref name="table"/>'s <paramref name="column"/> is indexed (one BOOLEAN row; none when missing).</summary>
/// <param name="table">The table name (prefix-validated, inlined — <c>table_columns()</c> takes a constant).</param>
/// <param name="column">The column name (a constant from <see cref="SymbolIndexes"/>).</param>
/// <returns>The query text.</returns>
public static string IndexedQuery(string table, string column) =>
$"SELECT indexed FROM table_columns('{table}') WHERE \"column\" = '{column}'";
/// <summary>
/// <c>ALTER TABLE t ALTER COLUMN tag ADD INDEX</c> — 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
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/QuestDbConformanceBackend.cs
+13 -1
View File
@@ -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);
}
/// <summary>Whether <paramref name="table"/>'s <paramref name="column"/> is indexed.</summary>
/// <param name="table">The table.</param>
/// <param name="column">The symbol column.</param>
/// <returns>The <c>table_columns()</c> flag.</returns>
public async Task<bool> 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());
}
/// <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()
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/TagIndexTests.cs
+36
View File
@@ -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<string> 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()
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbTablesTests.cs
+17 -2
View File
@@ -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");