feat(historian): Part 11 modified-value history on SqlServer + QuestDB (#668)

Every HistoryUpdate a SQL backend applies now leaves a modification trail
(prior value, principal, Part 11 update type, modification time), and
HistoryReadRaw with IsReadModified=true serves it as HistoryModifiedData.

- Core: HistoryUpdateContext threaded into all three IHistorianDataUpdater
  methods; IHistorianDataSource.ReadModifiedAsync (default NotSupported =>
  BadHistoryOperationUnsupported on Gateway/Null); HistoryModified* contracts.
- SqlServer: schema v2 (002_sample_modified.sql, RequiredVersion=2, docs copy);
  trail written by OUTPUT ... INTO on the same MERGE/DELETE statement, so a
  failed update leaves no audit row; retention purges SampleModified per tag on
  the source-time cutoff; ReadModifiedAsync keyed on (SourceTimeUtc, ModifiedId).
- QuestDB: append-only {prefix}sample_modified (TTL like samples), trail row
  written BEFORE the change (orphan-on-crash, never an untracked change);
  existence reads now carry the prior value; ReadModifiedAsync.
- Node manager: modified reads dispatch + page via the #669 continuation store
  (HistoryContinuationState gains HistoryCursorKind + LastModifiedId), gated by
  the existing HistoryRead decision; session principal / key id passed to the
  updater. IsDeleteModified stays unsupported.
- Client.CLI: historyread --modified.
- Tests: node-manager dispatch/paging/cursor binding/principal threading,
  over-the-wire HistoryModifiedData, shared ModifiedValueConformance on both
  backends, SqlServer atomicity (forced trail failure => no change) + v1->v2
  upgrade + retention trail purge.
- Docs: Historian.md (Modified-value history section, schema v2 upgrade note),
  security.md, Client.CLI.md, CLAUDE.md.
This commit is contained in:
Joseph Doherty
2026-10-11 09:26:36 -04:00
parent d803e0eab4
commit cfd403710e
51 changed files with 2173 additions and 195 deletions
@@ -22,6 +22,11 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// is written.</para>
/// <para><b>Unknown tag.</b> Insert/Update create it via <see cref="SqlServerTagCatalog"/>; Replace and the
/// deletes report <c>BadNoEntryExists</c> (DeleteRaw: <c>BadNoData</c>) without creating anything.</para>
/// <para><b>Modified-value history (Part 11, #668).</b> Every row the MERGE inserts or updates and every row a
/// delete removes is copied into <c>SampleModified</c> by an <c>OUTPUT … INTO</c> on the SAME statement — the prior
/// value from <c>deleted.*</c> (NULL for an insert), the caller's <see cref="HistoryUpdateContext"/>, the
/// <see cref="HistoryModificationType"/> and <c>SYSUTCDATETIME()</c>. One statement means the sample change and its
/// audit row commit or roll back together: a failed update leaves no audit row.</para>
/// </summary>
internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
{
@@ -39,10 +44,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> UpdateDataAsync(string tag, HistoryUpdateMode mode,
IReadOnlyList<HistorizationValue> values, CancellationToken ct)
IReadOnlyList<HistorizationValue> values, HistoryUpdateContext context, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(values);
ArgumentNullException.ThrowIfNull(context);
var plan = HistoryUpdatePlanner.Plan(tag, mode, values);
if (plan.Winners.Count == 0) return HistoryUpdatePlanner.Complete(plan, mode, new Dictionary<long, UpsertOutcome>());
@@ -52,14 +58,16 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
var outcome = tagId is null
? new Dictionary<long, UpsertOutcome>()
: await MergeAsync(tagId.Value, mode, plan.Winners.Select(w => w.Row).ToList(), ct).ConfigureAwait(false);
: await MergeAsync(tagId.Value, mode, plan.Winners.Select(w => w.Row).ToList(), context, ct).ConfigureAwait(false);
return HistoryUpdatePlanner.Complete(plan, mode, outcome);
}
/// <inheritdoc/>
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, CancellationToken ct)
public async Task<uint> DeleteRawAsync(string tag, DateTime startUtc, DateTime endUtc, HistoryUpdateContext context,
CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(context);
if (!HistorianTagNames.IsValid(tag)) return HistorianStatusCodes.BadInvalidArgument;
var tagId = await _tags.TryGetTagIdAsync(tag, ct).ConfigureAwait(false);
if (tagId is null) return HistoryUpdatePlanner.DeleteRawResult(tagKnown: false, 0);
@@ -67,9 +75,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
int rows;
await using (var cmd = _conn.CreateCommand(conn,
$"DELETE FROM {_conn.Q("Sample")} WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
$"DELETE FROM {_conn.Q("Sample")} {AuditInto(HistoryModificationType.Delete)} " +
"WHERE TagId = @tagId AND SourceTimeUtc >= @s AND SourceTimeUtc < @e", tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
AddAuditParameters(cmd, context);
SqlHistorianConnection.AddDateTime2(cmd, "@s", startUtc);
SqlHistorianConnection.AddDateTime2(cmd, "@e", endUtc);
rows = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
@@ -79,10 +89,12 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
}
/// <inheritdoc/>
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc, CancellationToken ct)
public async Task<IReadOnlyList<uint>> DeleteAtTimeAsync(string tag, IReadOnlyList<DateTime> timesUtc,
HistoryUpdateContext context, CancellationToken ct)
{
ArgumentNullException.ThrowIfNull(tag);
ArgumentNullException.ThrowIfNull(timesUtc);
ArgumentNullException.ThrowIfNull(context);
if (!HistorianTagNames.IsValid(tag))
{
var refused = new uint[timesUtc.Count];
@@ -100,9 +112,11 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
await using var tx = (SqlTransaction)await conn.BeginTransactionAsync(ct).ConfigureAwait(false);
await using (var cmd = _conn.CreateCommand(conn,
$"DELETE s OUTPUT deleted.SourceTimeUtc FROM {_conn.Q("Sample")} AS s JOIN @times AS t ON s.SourceTimeUtc = t.T WHERE s.TagId = @tagId", tx))
$"DELETE s {AuditInto(HistoryModificationType.Delete)} OUTPUT deleted.SourceTimeUtc " +
$"FROM {_conn.Q("Sample")} AS s JOIN @times AS t ON s.SourceTimeUtc = t.T WHERE s.TagId = @tagId", tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId.Value });
AddAuditParameters(cmd, context);
_conn.AddStructured(cmd, "@times", "TimeRow",
SampleRowMapper.ToTimeRecords(distinct.Select(t => new DateTime(t, DateTimeKind.Utc))));
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
@@ -114,7 +128,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
}
private async Task<Dictionary<long, UpsertOutcome>> MergeAsync(int tagId, HistoryUpdateMode mode,
List<HistorianSampleRow> rows, CancellationToken ct)
List<HistorianSampleRow> rows, HistoryUpdateContext context, CancellationToken ct)
{
var sql = $"MERGE {_conn.Q("Sample")} WITH (HOLDLOCK) AS t USING @rows AS r ON t.TagId = @tagId AND t.SourceTimeUtc = r.SourceTimeUtc ";
if (mode != HistoryUpdateMode.Insert)
@@ -123,7 +137,9 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
if (mode != HistoryUpdateMode.Replace)
sql += "WHEN NOT MATCHED THEN INSERT (TagId, SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc) " +
"VALUES (@tagId, r.SourceTimeUtc, r.ValueKind, r.NumValue, r.IntValue, r.StrValue, r.StatusCode, SYSUTCDATETIME()) ";
sql += "OUTPUT $action, r.SourceTimeUtc;";
// The MERGE's deleted.* is NULL for an inserted row, so one OUTPUT INTO records both an insert (no prior value)
// and a replace (the prior value). TagId comes from inserted.* — present on both branches.
sql += $"{AuditInto(ToModificationType(mode), "inserted")} OUTPUT $action, r.SourceTimeUtc;";
var outcome = new Dictionary<long, UpsertOutcome>();
await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
@@ -131,6 +147,7 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await using (var cmd = _conn.CreateCommand(conn, sql, tx))
{
cmd.Parameters.Add(new SqlParameter("@tagId", System.Data.SqlDbType.Int) { Value = tagId });
AddAuditParameters(cmd, context);
_conn.AddStructured(cmd, "@rows", "SampleRow", SampleRowMapper.ToSampleRecords(rows));
await using var rd = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
while (await rd.ReadAsync(ct).ConfigureAwait(false))
@@ -141,4 +158,33 @@ internal sealed class SqlServerHistorianDataUpdater : IHistorianDataUpdater
await tx.CommitAsync(ct).ConfigureAwait(false);
return outcome;
}
/// <summary>The Part 11 modification type recorded for an update mode.</summary>
/// <param name="mode">The update mode.</param>
/// <returns>Insert ⇒ Insert, Replace ⇒ Replace, Update ⇒ Update.</returns>
internal static HistoryModificationType ToModificationType(HistoryUpdateMode mode) => mode switch
{
HistoryUpdateMode.Insert => HistoryModificationType.Insert,
HistoryUpdateMode.Replace => HistoryModificationType.Replace,
_ => HistoryModificationType.Update,
};
/// <summary>
/// The <c>OUTPUT … INTO SampleModified</c> clause copying each affected row's prior value (<c>deleted.*</c>) into
/// the modification trail. <paramref name="keySource"/> names the pseudo-table that carries <c>TagId</c> /
/// <c>SourceTimeUtc</c> for every affected row (<c>deleted</c> for a DELETE, <c>inserted</c> for the MERGE).
/// </summary>
private string AuditInto(HistoryModificationType type, string keySource = "deleted") =>
$"OUTPUT {keySource}.TagId, {keySource}.SourceTimeUtc, SYSUTCDATETIME(), {(byte)type}, @principal, @apiKeyId, " +
"deleted.ValueKind, deleted.NumValue, deleted.IntValue, deleted.StrValue, deleted.StatusCode, deleted.ServerTimeUtc " +
$"INTO {_conn.Q("SampleModified")} (TagId, SourceTimeUtc, ModificationTimeUtc, UpdateType, Principal, ApiKeyId, " +
"ValueKind, NumValue, IntValue, StrValue, StatusCode, ServerTimeUtc)";
private static void AddAuditParameters(SqlCommand cmd, HistoryUpdateContext context)
{
cmd.Parameters.Add(new SqlParameter("@principal", System.Data.SqlDbType.NVarChar, HistoryUpdateContext.MaxIdentityLength)
{ Value = context.StoredPrincipal });
cmd.Parameters.Add(new SqlParameter("@apiKeyId", System.Data.SqlDbType.NVarChar, HistoryUpdateContext.MaxIdentityLength)
{ Value = (object?)context.StoredApiKeyId ?? DBNull.Value });
}
}