perf(historian): index the QuestDB sample tag column; read the data range before the buckets (#683)
QuestDB: {p}sample.tag is now SYMBOL INDEX. The processed-read data-range
lookups (EarliestSample / LatestSample) were partition scans of the unindexed
symbol; on 53.6M rows / 120 tags / 61 daily partitions the indexed seeks are
1-1.5 ms warm / 2-4 ms cold whatever the tag's history (unindexed: up to 11 ms
warm, 14 ms cold; cost grew with the table). The bootstrapper adds the index to a
pre-#683 table (ALTER ... ADD INDEX, a WAL txn, idempotent, ~3 s for 68M rows);
AutoCreateSchema=false only logs the statement. Disk +~12%; write apply time
within noise; main per-sample/raw/time-weighted queries unchanged or faster.
Rejected by measurement/analysis: bounded existence probes (still full scans
when a side is absent), LATEST ON (slowest unindexed), per-tag first/last side
table (cannot track TTL drops or deletes INSERT-only).
Both backends now read the data range before the bucket statement: an append
racing the read reaches the buckets but not the range, so it can no longer
produce a Good Count of 0 for an interval that has samples.
Tests: TagIndexTests (new-table index, bound-param seek uses the index,
tombstones excluded / outage marker counted, pre-#683 migration with identical
processed results across 10 aggregates x 4 windows x both directions).
This commit is contained in:
Joseph Doherty
+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);
|
||||
|
||||
+34
-1
@@ -1,3 +1,4 @@
|
||||
using System.Globalization;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Npgsql;
|
||||
@@ -15,10 +16,15 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
|
||||
/// <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.</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 +69,12 @@ internal sealed class QuestDbSchemaBootstrapper
|
||||
{
|
||||
await CreateTablesAsync(conn, ct).ConfigureAwait(false);
|
||||
await ReconcileTtlAsync(conn, ct).ConfigureAwait(false);
|
||||
await EnsureTagIndexAsync(conn, apply: true, ct).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
await VerifyAsync(conn, ct).ConfigureAwait(false);
|
||||
await EnsureTagIndexAsync(conn, apply: false, ct).ConfigureAwait(false);
|
||||
}
|
||||
_ready = true;
|
||||
_failure.Clear();
|
||||
@@ -117,6 +125,31 @@ internal sealed class QuestDbSchemaBootstrapper
|
||||
}
|
||||
}
|
||||
|
||||
/// <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.
|
||||
await using (var alter = _conn.CreateCommand(conn, QuestDbTables.AddTagIndex(table)))
|
||||
await alter.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
|
||||
_log.LogInformation("Added the tag index to QuestDB table {Table} (#683).", table);
|
||||
}
|
||||
|
||||
private async Task ReconcileTtlAsync(NpgsqlConnection conn, CancellationToken ct)
|
||||
{
|
||||
var wantHours = _options.RetentionDays is { } d ? d * 24L : 0L;
|
||||
|
||||
@@ -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";
|
||||
|
||||
+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);
|
||||
|
||||
Reference in New Issue
Block a user