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

feat(historian): QuestDB HistoryUpdate with soft delete and read-your-writes

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 02:04:48 -04:00
parent ecc0f914cc
commit ee174c0179
3 changed files with 543 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataUpdater.cs
+258
View File
@@ -0,0 +1,258 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Npgsql;
using NpgsqlTypes;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb;
/// <summary>
/// OPC UA HistoryUpdate against the QuestDB sample table (spec §2.3). Every operation is an <c>INSERT</c>: the
/// table's <c>DEDUP UPSERT KEYS(ts, tag)</c> performs the replace, and a delete writes a tombstone
/// (<c>deleted = true</c>, NULL values, <c>BadNoData</c>). QuestDB <c>UPDATE</c>/<c>DELETE</c> are never issued.
/// <para>Every status comes from <see cref="HistoryUpdatePlanner"/>, so codes equal the SQL Server backend's.
/// "Exists" means present and not deleted, so Insert over a tombstone succeeds (<c>GoodEntryInserted</c>) while
/// Replace and DeleteAtTime treat a tombstone as no entry. Existence is read with one range query and filtered
/// client-side (never an IN list).</para>
/// <para><b>Pre-1970 (RC1):</b> QuestDB cannot store such a designated timestamp. Insert/Replace/Update of such an
/// item reports <c>BadOutOfRange</c> and sends nothing (a null/<c>MinValue</c> timestamp is still
/// <c>BadInvalidTimestamp</c>); DeleteAtTime reports <c>BadNoEntryExists</c>; a DeleteRaw window may start before
/// 1970 (read bounds are not clamped).</para>
/// <para><b>Read-your-writes:</b> after writing, <see cref="QuestDbWalVisibility"/> waits for the rows to become
/// visible, bounded by <c>VisibilityTimeout</c>; a timeout only logs a Warning, never changes a code.</para>
/// </summary>
internal sealed class QuestDbHistorianDataUpdater : IHistorianDataUpdater
{
/// <summary>Rows per INSERT and per DeleteRaw page (T0 Q2: 9,000 parameters, far under the 32,767 limit).</summary>
internal const int BatchSize = 1000;
private const long ResolutionTicks = 10;
private readonly QuestDbConnection _conn;
private readonly QuestDbTagCatalog _tags;
private readonly QuestDbWalVisibility _wal;
private readonly ILogger _log;
/// <summary>Creates the updater.</summary>
/// <param name="conn">Connection helper.</param>
/// <param name="tags">Tag catalog (R7: the only known-tag authority).</param>
/// <param name="wal">WAL visibility wait (R5).</param>
/// <param name="log">Optional logger.</param>
public QuestDbHistorianDataUpdater(QuestDbConnection conn, QuestDbTagCatalog tags, QuestDbWalVisibility wal,
ILogger? log = null)
{
_conn = conn;
_tags = tags;
_wal = wal;
_log = log ?? NullLogger.Instance;
}
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList<HistorizationValue> values, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(values);
// RC1: a valid-timestamp item before 1970 can never be stored; refuse it and plan the rest.
var final = new uint[values.Count];
var kept = new List<int>(values.Count);
for (var i = 0; i < values.Count; i++)
{
if (values[i].TimestampUtc is { } ts && ts != DateTime.MinValue && !QuestDbConnection.IsStorable(ts))
final[i] = HistorianStatusCodes.BadOutOfRange;
else
kept.Add(i);
}
var plan = HistoryUpdatePlanner.Plan(tag, mode, kept.Select(i => values[i]).ToList(), ResolutionTicks);
IReadOnlyList<uint> planned;
if (plan.Winners.Count == 0)
{
planned = HistoryUpdatePlanner.Complete(plan, mode, new Dictionary<long, UpsertOutcome>());
}
else
{
var known = mode == HistoryUpdateMode.Replace
? await _tags.ExistsAsync(tag, ct).ConfigureAwait(false)
: await EnsureAsync(tag, plan.Winners[0].Row.Value.Kind, ct).ConfigureAwait(false);
var outcome = known
? await ApplyAsync(tag, mode, plan.Winners.Select(w => w.Row).ToList(), ct).ConfigureAwait(false)
: new Dictionary<long, UpsertOutcome>();
planned = HistoryUpdatePlanner.Complete(plan, mode, outcome);
}
for (var k = 0; k < kept.Count; k++) final[kept[k]] = planned[k];
return final;
}
/// <inheritdoc/>
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument;
if (!await _tags.ExistsAsync(tag, ct).ConfigureAwait(false))
return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0);
var cursor = QuestDbConnection.Round(startUtc);
var end = QuestDbConnection.Round(endUtc);
long total = 0;
while (cursor < end)
{
var page = new List<DateTime>(BatchSize);
await using (var conn = await _conn.OpenAsync(ct).ConfigureAwait(false))
await using (var cmd = _conn.CreateCommand(conn,
$"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts LIMIT {BatchSize}"))
{
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindTimestamp(cmd, cursor);
QuestDbConnection.BindTimestamp(cmd, end);
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false)) page.Add(QuestDbConnection.ReadUtc(rd, 0));
}
if (page.Count == 0) break;
await WriteTombstonesAsync(tag, page, 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;
if (page.Count < BatchSize) break;
cursor = page[^1].AddTicks(ResolutionTicks);
}
return HistoryUpdatePlanner.DeleteRawResult(tagKnown: true, total);
}
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(timesUtc);
if (!HistorianTagNames.IsValid(tag))
{
var refused = new uint[timesUtc.Count];
Array.Fill(refused, HistorianStatusCodes.BadInvalidArgument);
return refused;
}
// M-2: round BEFORE both the delete and the comparison, so a sub-µs request time matches the row it deleted.
// MinValue stays as is: the planner turns it into BadInvalidTimestamp.
var rounded = timesUtc.Select(t => t == DateTime.MinValue ? t : QuestDbConnection.Round(t)).ToList();
var candidates = new HashSet<long>();
foreach (var t in rounded)
if (t != DateTime.MinValue && t >= DateTime.UnixEpoch) candidates.Add(t.Ticks); // pre-1970 can hold no row
var deleted = new HashSet<long>();
if (candidates.Count == 0 || !await _tags.ExistsAsync(tag, ct).ConfigureAwait(false))
return HistoryUpdatePlanner.DeleteAtTimeResults(rounded, deleted, tagKnown: false);
var live = await ReadLiveAsync(tag, candidates.Min(), candidates.Max(), ct).ConfigureAwait(false);
var toDelete = candidates.Where(live.Contains).OrderBy(x => x).ToList();
foreach (var chunk in toDelete.Chunk(BatchSize))
await WriteTombstonesAsync(tag, chunk.Select(t => new DateTime(t, DateTimeKind.Utc)).ToList(), ct).ConfigureAwait(false);
if (toDelete.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
deleted.UnionWith(toDelete);
return HistoryUpdatePlanner.DeleteAtTimeResults(rounded, deleted, tagKnown: true);
}
private async Task<bool> EnsureAsync(string tag, HistorianValueKind kind, CancellationToken ct)
{
await _tags.EnsureAsync(tag, HistorianDataTypes.FromKind(kind), ct).ConfigureAwait(false);
return true;
}
private async Task<Dictionary<long, UpsertOutcome>> ApplyAsync(string tag, HistoryUpdateMode mode,
List<HistorianSampleRow> rows, CancellationToken ct)
{
// Key = the planner's tick (what Complete looks up); the stored time is Round(of the same row).
var live = await ReadLiveAsync(tag, rows.Min(r => QuestDbConnection.Round(r.TimeUtc).Ticks),
rows.Max(r => QuestDbConnection.Round(r.TimeUtc).Ticks), ct).ConfigureAwait(false);
var outcome = new Dictionary<long, UpsertOutcome>();
var toWrite = new List<HistorianSampleRow>();
foreach (var r in rows)
{
var present = live.Contains(QuestDbConnection.Round(r.TimeUtc).Ticks);
if (mode == HistoryUpdateMode.Insert && present) continue;
if (mode == HistoryUpdateMode.Replace && !present) continue;
outcome[r.TimeUtc.Ticks] = present ? UpsertOutcome.Replaced : UpsertOutcome.Inserted;
toWrite.Add(r);
}
foreach (var chunk in toWrite.Chunk(BatchSize))
await WriteRowsAsync(tag, chunk, ct).ConfigureAwait(false);
if (toWrite.Count > 0) await _wal.WaitAsync(_conn.SampleTable, ct).ConfigureAwait(false);
return outcome;
}
/// <summary>Ticks of the live (present, not deleted) rows with <c>ts</c> in <c>[minTicks, maxTicks]</c>: one range query.</summary>
private async Task<HashSet<long>> ReadLiveAsync(string tag, long minTicks, long maxTicks, CancellationToken ct)
{
var live = new HashSet<long>();
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn,
$"SELECT ts FROM {_conn.SampleTable} WHERE tag = $1 AND ts >= $2 AND ts <= $3 AND NOT deleted");
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindTimestamp(cmd, new DateTime(minTicks, DateTimeKind.Utc));
QuestDbConnection.BindTimestamp(cmd, new DateTime(maxTicks, DateTimeKind.Utc));
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false)) live.Add(QuestDbConnection.ReadUtc(rd, 0).Ticks);
return live;
}
private Task WriteRowsAsync(string tag, IReadOnlyList<HistorianSampleRow> rows, CancellationToken ct) =>
InsertAsync(rows.Count, ct, (cmd, i) =>
{
var row = rows[i];
var c = SampleColumnMapper.ToColumns(row.Value);
QuestDbConnection.BindTimestamp(cmd, row.TimeUtc, clampToEpoch: true);
QuestDbConnection.BindText(cmd, tag);
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(row.StatusCode));
Typed(cmd, NpgsqlDbType.Boolean, false);
});
private Task WriteTombstonesAsync(string tag, IReadOnlyList<DateTime> times, CancellationToken ct) =>
InsertAsync(times.Count, ct, (cmd, i) =>
{
QuestDbConnection.BindTimestamp(cmd, times[i], clampToEpoch: true);
QuestDbConnection.BindText(cmd, tag);
QuestDbConnection.BindByte(cmd, (byte)HistorianValueKind.Empty);
Typed(cmd, NpgsqlDbType.Double, null);
Typed(cmd, NpgsqlDbType.Bigint, null);
Typed(cmd, NpgsqlDbType.Varchar, null);
QuestDbConnection.BindLong(cmd, SampleColumnMapper.ToLongStatus(HistorianStatusCodes.BadNoData));
Typed(cmd, NpgsqlDbType.Boolean, true);
});
/// <summary>Multi-row INSERT; <paramref name="bindRow"/> adds the first eight parameters of a row, this adds server_ts.</summary>
private async Task InsertAsync(int count, CancellationToken ct, Action<NpgsqlCommand, int> bindRow)
{
if (count == 0) return;
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
await using var cmd = _conn.CreateCommand(conn, "");
var now = DateTime.UtcNow;
var sb = new System.Text.StringBuilder(
$"INSERT INTO {_conn.SampleTable} (ts, tag, kind, num_value, int_value, str_value, status, deleted, server_ts) VALUES ");
for (var i = 0; i < count; i++)
{
bindRow(cmd, i);
QuestDbConnection.BindTimestamp(cmd, now);
var b = i * 9;
if (i > 0) sb.Append(',');
sb.Append('(');
for (var p = 1; p <= 9; p++) sb.Append(p > 1 ? "," : "").Append('$').Append(b + p);
sb.Append(')');
}
cmd.CommandText = sb.ToString();
try
{
await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
_log.LogWarning(ex, "QuestDB HistoryUpdate INSERT of {Count} rows into {Table} failed.", count, _conn.SampleTable);
throw;
}
}
private static void Typed(NpgsqlCommand cmd, NpgsqlDbType type, object? value) =>
cmd.Parameters.Add(new NpgsqlParameter { NpgsqlDbType = type, Value = value ?? DBNull.Value });
}