IStreamDataRepository
Namespace: Meshmakers.Octo.Runtime.Contracts.StreamData
Repository for stream data operations against a tenant-scoped, CrateDB-backed time-series store
organised by CkArchive instances. Accessed via
ITenantContext.GetStreamDataRepository(). Every operation is scoped to a single archive,
addressed by its runtime id.
public interface IStreamDataRepository
Methods
EnsureDatabaseCreatedAsync()
Ensures the tenant's stream data namespace (e.g. CrateDB schema) exists. Idempotent. Called when a tenant opts into stream data; per-archive tables are created later via IStreamDataRepository.EnsureArchiveCreatedAsync(ArchiveSnapshot).
Task EnsureDatabaseCreatedAsync()
Returns
DeleteDatabaseAsync()
Drops the tenant's entire stream data namespace including every archive table. Idempotent.
Task DeleteDatabaseAsync()
Returns
EnsureArchiveCreatedAsync(ArchiveSnapshot)
Creates the storage table for the archive described by snapshot according
to its current CkArchive definition. The snapshot carries the target CK type and
user-picked columns the data store needs to generate DDL — passing it directly avoids a
round-trip through IArchiveRuntimeStore from inside the repository.
Idempotent (uses CREATE TABLE IF NOT EXISTS) so retries after a transient Mongo
update failure converge cleanly.
Task EnsureArchiveCreatedAsync(ArchiveSnapshot snapshot)
Parameters
snapshot ArchiveSnapshot
Returns
DeleteArchiveAsync(OctoObjectId)
Drops the storage table for the archive identified by archiveRtId.
Idempotent. Called from the lifecycle service when an archive is deleted.
Task DeleteArchiveAsync(OctoObjectId archiveRtId)
Parameters
archiveRtId OctoObjectId
Returns
ValidateComputedColumnsAsync(OctoObjectId, IReadOnlyList<CkArchiveColumnSpec>, CancellationToken)
Validates a prospective computed-column set (AB#4189 Phase 7) — syntax, reference resolution,
acyclic dependencies, Path/Formula exclusivity, nullable, ResultType — without touching
storage. The lifecycle service calls this before persisting an add (existing columns + the
new one) or a remove (existing columns − the removed one, so a now-dangling reference is
rejected). Throws a StreamDataException (stable GraphQL error code) on the first issue.
Task ValidateComputedColumnsAsync(OctoObjectId archiveRtId, IReadOnlyList<CkArchiveColumnSpec> prospectiveColumns, CancellationToken cancellationToken)
Parameters
archiveRtId OctoObjectId
prospectiveColumns IReadOnlyList<CkArchiveColumnSpec>
cancellationToken CancellationToken
Returns
NormalizeComputedFormulaAsync(OctoObjectId, IReadOnlyList<CkArchiveColumnSpec>, String, CancellationToken)
Rewrites a computed-column formula from the archive's logical column vocabulary — the CK
attribute paths the Studio lists and the query surface uses, e.g. Amount.Value — into the
physical form the evaluation path binds, e.g. amountvalue (AB#4779).
The lifecycle service calls this before validating and persisting, so the stored formula is always physical and every downstream consumer keeps seeing exactly what it saw before. A name that is already physical, or that matches no column at all, is left as written — the latter so validation can reject it by the spelling the caller actually used.
It lives on the storage contract rather than in the lifecycle service because the logical→physical rule belongs to the storage layer, the same reason IStreamDataRepository.ValidateComputedColumnsAsync(OctoObjectId, IReadOnlyList<CkArchiveColumnSpec>, CancellationToken) does. No I/O.
Task<string> NormalizeComputedFormulaAsync(OctoObjectId archiveRtId, IReadOnlyList<CkArchiveColumnSpec> columns, string formula, CancellationToken cancellationToken)
Parameters
archiveRtId OctoObjectId
columns IReadOnlyList<CkArchiveColumnSpec>
formula String
cancellationToken CancellationToken
Returns
AddComputedColumnStorageAsync(ArchiveSnapshot, String, CancellationToken)
Adds the physical CrateDB column for the computed column named columnName
via ALTER TABLE … ADD COLUMN (AB#4189 Phase 7). Idempotent — tolerant of the column
already existing (e.g. a re-add reusing an orphaned column). The column on
snapshot supplies the declared result type the physical column is typed from.
Task AddComputedColumnStorageAsync(ArchiveSnapshot snapshot, string columnName, CancellationToken cancellationToken)
Parameters
snapshot ArchiveSnapshot
columnName String
cancellationToken CancellationToken
Returns
AddPendingComputedColumnStorageAsync(ArchiveSnapshot, String, CancellationToken)
Adds the physical CrateDB column for the pending version of a computed column
({base}__v{ComputedVersion+1}) via ALTER TABLE … ADD COLUMN (AB#4189 Phase 7,
formula change). Idempotent. Must run before the pending formula is marked so ingest's
dual-write never targets a missing column.
Task AddPendingComputedColumnStorageAsync(ArchiveSnapshot snapshot, string columnName, CancellationToken cancellationToken)
Parameters
snapshot ArchiveSnapshot
columnName String
cancellationToken CancellationToken
Returns
BackfillComputedColumnAsync(ArchiveSnapshot, String, CancellationToken)
Backfills the computed column named columnName across the archive's
existing rows (AB#4189 Phase 7, §8): pages through the rows, evaluates the column's formula
per row, and writes the result into its physical cell. The physical column must already exist
(see IStreamDataRepository.AddComputedColumnStorageAsync(ArchiveSnapshot, String, CancellationToken)). Readers keep seeing the previous state until
the lifecycle flips the column to Active, because the column is hidden while
non-active. Throws on a storage failure so the lifecycle can mark the column Failed.
Task BackfillComputedColumnAsync(ArchiveSnapshot snapshot, string columnName, CancellationToken cancellationToken)
Parameters
snapshot ArchiveSnapshot
columnName String
cancellationToken CancellationToken
Returns
InsertAsync(OctoObjectId, StreamDataPoint)
Inserts a single data point into the archive. Throws
ArchiveNotActivatedException if the archive's status is not Activated;
throws RequiredAttributeMissingException on a missing required path.
Task InsertAsync(OctoObjectId archiveRtId, StreamDataPoint datapoint)
Parameters
archiveRtId OctoObjectId
datapoint StreamDataPoint
Returns
InsertAsync(OctoObjectId, IEnumerable<StreamDataPoint>)
Inserts multiple data points into the archive. Pre-validates the entire batch before any SQL is sent; on first violation no row is written and the offending point's index is surfaced via the thrown exception.
Task InsertAsync(OctoObjectId archiveRtId, IEnumerable<StreamDataPoint> datapoints)
Parameters
archiveRtId OctoObjectId
datapoints IEnumerable<StreamDataPoint>
Returns
InsertTimeRangeAsync(OctoObjectId, IEnumerable<TimeRangeStreamDataPoint>, CancellationToken)
Inserts externally pre-aggregated time-range data points into a TimeRangeArchive.
Each point carries an explicit [from, to) window; the natural key
(window_start, window_end, rtid, ckTypeId) handles re-deliveries via
ON CONFLICT DO UPDATE, setting the row's was_updated flag to true on every
upsert. Concept §3 / §5. Throws ArchiveNotActivatedException if the archive is not
in Activated state, and ArgumentException if any point has To <= From.
Task InsertTimeRangeAsync(OctoObjectId archiveRtId, IEnumerable<TimeRangeStreamDataPoint> datapoints, CancellationToken cancellationToken)
Parameters
archiveRtId OctoObjectId
datapoints IEnumerable<TimeRangeStreamDataPoint>
cancellationToken CancellationToken
Returns
ExecuteQueryAsync(OctoObjectId, StreamDataQueryOptions)
Executes a simple stream data query against the archive.
Task<StreamDataQueryResult> ExecuteQueryAsync(OctoObjectId archiveRtId, StreamDataQueryOptions options)
Parameters
archiveRtId OctoObjectId
options StreamDataQueryOptions
Returns
ExecuteAggregationQueryAsync(OctoObjectId, StreamDataAggregationQueryOptions)
Executes an aggregation query (without grouping) against the archive.
Task<StreamDataQueryResult> ExecuteAggregationQueryAsync(OctoObjectId archiveRtId, StreamDataAggregationQueryOptions options)
Parameters
archiveRtId OctoObjectId
options StreamDataAggregationQueryOptions
Returns
ExecuteGroupedAggregationQueryAsync(OctoObjectId, StreamDataGroupedAggregationQueryOptions)
Executes a grouped aggregation query against the archive.
Task<StreamDataQueryResult> ExecuteGroupedAggregationQueryAsync(OctoObjectId archiveRtId, StreamDataGroupedAggregationQueryOptions options)
Parameters
archiveRtId OctoObjectId
options StreamDataGroupedAggregationQueryOptions
Returns
ExecuteDownsamplingQueryAsync(OctoObjectId, StreamDataDownsamplingQueryOptions)
Executes a downsampling query with time bins against the archive.
Task<StreamDataQueryResult> ExecuteDownsamplingQueryAsync(OctoObjectId archiveRtId, StreamDataDownsamplingQueryOptions options)
Parameters
archiveRtId OctoObjectId
options StreamDataDownsamplingQueryOptions
Returns
ExportRowsAsync(OctoObjectId, TimeWindow, CancellationToken)
Streams the rows of the archive in a stable key order (keyset pagination on the natural
key) so the caller can serialise NDJSON without buffering the table. When
window is non-null only rows whose timestamp (raw) or
window_start (windowed: rollup / time-range) fall in [FromUtc, ToUtc) are
emitted — the predicate rides on the already time-ordered keyset scan, so a windowed export
is no more expensive than a full one (and cheaper). Each row is yielded as a dictionary of
physical CrateDB column name → value (the standard columns plus the user columns). An archive
without a backing table (e.g. Created) yields no rows rather than throwing.
Archive data export/import concept (AB#4230) §4.1.
IAsyncEnumerable<IReadOnlyDictionary<string, object>> ExportRowsAsync(OctoObjectId archiveRtId, TimeWindow window, CancellationToken ct)
Parameters
archiveRtId OctoObjectId
window TimeWindow
Returns
IAsyncEnumerable<IReadOnlyDictionary<String, Object>>
ImportRowsAsync(OctoObjectId, IAsyncEnumerable<IReadOnlyDictionary<String, Object>>, ArchiveImportMode, CancellationToken)
Bulk-inserts pre-parsed archive rows (the inverse of IStreamDataRepository.ExportRowsAsync(OctoObjectId, TimeWindow, CancellationToken)). Rows are
streamed in batches into the existing batched insert path (the single-timestamp insert for
raw archives, the (window_start, window_end) ON CONFLICT path for windowed archives).
mode selects insert-only versus upsert semantics on the natural key. Each
row's rtid is validated as 24-char hex; a violation surfaces a per-field error rather
than a generic message. Archive data export/import concept (AB#4230) §4.1 / §7 / §10.
Task ImportRowsAsync(OctoObjectId archiveRtId, IAsyncEnumerable<IReadOnlyDictionary<string, object>> rows, ArchiveImportMode mode, CancellationToken ct)
Parameters
archiveRtId OctoObjectId
rows IAsyncEnumerable<IReadOnlyDictionary<String, Object>>
mode ArchiveImportMode
Returns
AggregateBucketAsync(ArchiveSnapshot, RollupArchiveSnapshot, DateTime, DateTime, CancellationToken)
Aggregates one bucket from sourceArchive into rollup:
reads source rows with timestamp ∈ [bucketStart, bucketEnd), groups by rtId,
applies the CkRollupAggregationSpec aggregations, and upserts one row per
entity into the rollup archive's table with timestamp = bucketEnd. Rollup-archives
concept §5.
Task<int> AggregateBucketAsync(ArchiveSnapshot sourceArchive, RollupArchiveSnapshot rollup, DateTime bucketStart, DateTime bucketEnd, CancellationToken cancellationToken)
Parameters
sourceArchive ArchiveSnapshot
rollup RollupArchiveSnapshot
bucketStart DateTime
bucketEnd DateTime
cancellationToken CancellationToken
Returns
Remarks:
Idempotent on the natural key (timestamp, rtId): when the same bucket is
re-aggregated (e.g. after a watermark rewind, or a crash before
IRollupArchiveRuntimeStore.AdvanceWatermarkAsync(OctoObjectId, DateTime, Boolean) committed), the
implementation must collapse duplicates via the data store's upsert primitive
(CrateDB: ON CONFLICT (timestamp, rtId) DO UPDATE) so the orchestrator can
always retry safely. Returns the number of upserted target rows.
Single-source per call, unchanged by AB#5157: a multi-source rollup aggregates one bucket from exactly one source, and the orchestrator picks that source per bucket via RollupArchiveSnapshot.SourceForBucket(DateTime, DateTime) before calling in. A bucket no source span covers is never passed here at all — the orchestrator writes no row and advances the watermark instead.
ClearRecomputeGenerationsAsync(OctoObjectId, DateTime, CancellationToken)
Clears the per-window recompute generation pointers for a rollup archive at and after
fromBucketEnd, so a subsequent forward re-aggregation (which writes
generation 0) becomes visible again. Called when a rollup watermark is rewound over a range
that was previously recomputed (AB#4184, Phase 6): the rewind re-aggregates forward at
generation 0, but the active-generation pointer would otherwise keep readers on the stale
recomputed generation. Implementations remove the generation-map entries whose range reaches
past fromBucketEnd and the superseded (generation > 0) rows in that
range. Idempotent — a no-op when there is no recompute state (no genmap / non-rollup).
Task ClearRecomputeGenerationsAsync(OctoObjectId rollupRtId, DateTime fromBucketEnd, CancellationToken cancellationToken)
Parameters
rollupRtId OctoObjectId
fromBucketEnd DateTime
cancellationToken CancellationToken
Returns
GetArchiveMinTimestampAsync(OctoObjectId, CancellationToken)
Resolves the earliest stored timestamp in the archive's backing table — MIN(window_start)
for windowed (rollup / time-range) archives, MIN(timestamp) for raw archives (AB#4269).
Used by rollup backfill to recompute a rollup over its source archive's entire history without
the operator supplying a start timestamp. Returns null when the archive has no backing
table yet (e.g. Created) or the table is empty.
Task<Nullable<DateTime>> GetArchiveMinTimestampAsync(OctoObjectId archiveRtId, CancellationToken cancellationToken)
Parameters
archiveRtId OctoObjectId
cancellationToken CancellationToken
Returns
GetArchiveCoverageAsync(OctoObjectId, CancellationToken)
Measures the archive's stored time range (AB#5157): MIN(window_start) /
MAX(window_end) for windowed (rollup / time-range) archives and
MIN(timestamp) / MAX(timestamp) for raw archives. Returns null when the
archive holds no data — no backing table yet (e.g. Created), an unknown relation, or
an empty table — never a sentinel range. Every other storage failure propagates.
Task<ArchiveCoverage> GetArchiveCoverageAsync(OctoObjectId archiveRtId, CancellationToken cancellationToken)
Parameters
archiveRtId OctoObjectId
cancellationToken CancellationToken
Returns
Remarks:
Coverage is archive-wide (not per entity rtId) and a single from/to pair; gaps inside the range are not represented. Callers normally go through IArchiveCoverageProvider so the answer is memoised per tenant instead of probed on every request.
GetArchiveStatsAsync(IReadOnlyList<OctoObjectId>, CancellationToken)
Returns per-archive storage stats (row count, on-disk size, health) for each
archiveRtIds entry. Bulk call so the studio's archives list can render
stats columns without an N+1 round-trip per row. Archives whose backing table doesn't
exist yet (not activated, or post-delete) appear in the result with
ArchiveStorageStats.TableExists false and zero counters — the caller is not
expected to filter the input list beforehand.
Task<IReadOnlyDictionary<OctoObjectId, ArchiveStorageStats>> GetArchiveStatsAsync(IReadOnlyList<OctoObjectId> archiveRtIds, CancellationToken cancellationToken)
Parameters
archiveRtIds IReadOnlyList<OctoObjectId>
cancellationToken CancellationToken
Returns
Task<IReadOnlyDictionary<OctoObjectId, ArchiveStorageStats>>
Remarks:
Implementations may issue a single underlying query against their introspection surface
(e.g. CrateDB sys.shards + sys.health) and return one entry per requested
rtId. Order of returned entries is implementation-defined; callers must look up by rtId.