diff --git a/internal/app/backend.go b/internal/app/backend.go index 1bea7fc..c62e8b5 100644 --- a/internal/app/backend.go +++ b/internal/app/backend.go @@ -502,7 +502,7 @@ func buildDuckLakeMaterializer(settings *config.Settings, pollInterval time.Dura dailyMode, ok := materializer.ParseDailyRollupMode(settings.MaterializerDailyRollupMode) if !ok { _ = duckSvc.Close() - return nil, nil, nil, fmt.Errorf("invalid MATERIALIZER_DAILY_ROLLUP_MODE %q (off|shadow)", settings.MaterializerDailyRollupMode) + return nil, nil, nil, fmt.Errorf("invalid MATERIALIZER_DAILY_ROLLUP_MODE %q (off|on; shadow retired in dq#55 step 5 — move to on)", settings.MaterializerDailyRollupMode) } var dailyDelay time.Duration if settings.MaterializerDailyRollupDelay != "" { diff --git a/internal/config/settings.go b/internal/config/settings.go index 4d38b3f..211f878 100644 --- a/internal/config/settings.go +++ b/internal/config/settings.go @@ -132,14 +132,14 @@ type Settings struct { // reachable from config. MaterializerMaxSnapshotSpan int `yaml:"MATERIALIZER_MAX_SNAPSHOT_SPAN"` // MaterializerDailyRollupMode gates the daily signals_latest refresh (dq#55): - // "off" (default) does nothing; "shadow" maintains lake.signals_latest_daily - // by a once-daily watermarked fold while the per-pass fold keeps maintaining - // lake.signals_latest, and diffs the two after each refresh; "on" is the - // step-4 flip — the daily refresh maintains lake.signals_latest itself, the - // per-pass fold is off, and the first boot after shadow PROMOTES the - // validated shadow table (discarding the fold-era table, duplicate - // corruption included). Pair "on" with LAKE_ROLLUP_DAILY_SERVING=true on - // the query fleet. Materializer-only. + // "on" (the default, also for empty) — the once-daily watermarked refresh is + // THE maintainer of lake.signals_latest (the per-pass fold was removed in + // step 5); "off" disables the refresh entirely and leaves the rollup + // UNMAINTAINED — tests/one-off ops only, warned at boot. The retired + // "shadow" value is now invalid: shadow-era configs must move to "on" (a + // leftover lake.signals_latest_daily table is promoted automatically at + // first boot). Pair "on" with LAKE_ROLLUP_DAILY_SERVING=true on the query + // fleet. Materializer-only. MaterializerDailyRollupMode string `yaml:"MATERIALIZER_DAILY_ROLLUP_MODE"` // MaterializerDailyRollupDelay is a Go duration: how long after the // UTC-midnight partition rollover the daily refresh waits before folding the @@ -197,10 +197,13 @@ type Settings struct { // not be off (the store connection rides on it). Query-fleet only. LatestKVReadModeExtended string `yaml:"LATEST_KV_READ_MODE_EXTENDED"` // LakeRollupDailyServing marks lake.signals_latest as maintained by the - // DAILY watermarked refresh (the dq#55 step-4 flip): summaries then serve - // the exact (rollup ∪ signals-since-watermark) union instead of the plain - // rollup read. MUST be false while the per-pass fold maintains the rollup - // (the union would double-count the tail). Query-fleet only; default false. + // DAILY watermarked refresh: summaries then serve the exact (rollup ∪ + // signals-since-watermark) union instead of the plain rollup read. With the + // per-pass fold removed (dq#55 step 5) the daily refresh is the rollup's + // only maintainer, so this belongs true on the query fleet whenever the + // materializer runs the default MATERIALIZER_DAILY_ROLLUP_MODE=on — a plain + // rollup read under-counts the post-watermark tail. Query-fleet only; + // default false only so a mode=off test/ops setup isn't unioned twice. LakeRollupDailyServing bool `yaml:"LAKE_ROLLUP_DAILY_SERVING"` // LatestKVForceBootstrap re-runs the lake.signals_latest → KV bootstrap on // boot even though the completion marker is present — the repair for a diff --git a/internal/latestkv/latestkv.go b/internal/latestkv/latestkv.go index 1e66a32..b24291d 100644 --- a/internal/latestkv/latestkv.go +++ b/internal/latestkv/latestkv.go @@ -8,7 +8,7 @@ // The bucket is a CACHE of the lake, never the source of truth: the writer // (the single materializer, via materializer.LatestPublisher) folds each // decoded batch in last-write-wins by (timestamp DESC, cloud_event_id ASC) — -// the exact recency order foldSignalsRollup uses — so publishes are idempotent +// the exact recency order rollupSelectSQL uses — so publishes are idempotent // under NATS redelivery, window replay, and backfill. A lost update (KV outage, // crash) heals per (subject, name) on that signal's next reading, or wholesale // via BootstrapFromRollup. Readers must treat a miss or an unreachable bucket @@ -142,7 +142,7 @@ func (e *Entry) LastSeen() time.Time { // newerThan reports whether (ts, ceid) beats (oldTS, oldCEID) under the // rollup's recency order: ORDER BY timestamp DESC, cloud_event_id ASC. On an // exact timestamp tie the LEXICOGRAPHICALLY SMALLER cloud_event_id wins — -// matching foldSignalsRollup/rollupSelectSQL so the KV and the rollup pick the +// matching rollupSelectSQL so the KV and the rollup pick the // same winner and the phase-2 fallback path can't flap between two values. func newerThan(ts time.Time, ceid string, oldTS time.Time, oldCEID string) bool { if ts.After(oldTS) { diff --git a/internal/latestkv/latestkv_test.go b/internal/latestkv/latestkv_test.go index d6763ee..e427bc3 100644 --- a/internal/latestkv/latestkv_test.go +++ b/internal/latestkv/latestkv_test.go @@ -32,7 +32,7 @@ func TestFold_NewerTimestampWins(t *testing.T) { } // The rollup breaks exact-timestamp ties by cloud_event_id ASC -// (foldSignalsRollup's QUALIFY ordering); the KV fold must pick the same +// (rollupSelectSQL's QUALIFY ordering); the KV fold must pick the same // winner so the phase-2 rollup fallback can't flap between two values. func TestFold_EqualTimestampSmallerCEIDWins(t *testing.T) { var e Entry diff --git a/internal/materializer/daily_rollup.go b/internal/materializer/daily_rollup.go index 42bf6dc..66ecdae 100644 --- a/internal/materializer/daily_rollup.go +++ b/internal/materializer/daily_rollup.go @@ -12,29 +12,38 @@ import ( "github.com/DIMO-Network/dq/internal/service/duck" ) -// The daily rollup refresh (dq#55, step 1). +// The daily rollup refresh (dq#55) — since step 5, THE maintenance path for +// lake.signals_latest. // -// The per-pass incremental fold (captureRollupDelta + foldSignalsRollup) keeps -// lake.signals_latest continuously exact, but its lake.signals scans carry a -// correlated bound (s.timestamp >= prev_ts) that cannot drive static partition -// pruning — every span re-opens roughly the day partition's current file set, -// three times, and the file count grows all day. That is the daily pass-duration -// sawtooth. Since the KV serves signalsLatest (LATEST_KV_READ_MODE=serve), the -// rollup's remaining jobs — KV bootstrap/reconcile source, rare fallback — have -// no freshness SLA of their own, so the plan is to refresh it once daily from a -// WATERMARK: a constant timestamp literal that prunes the fold to the settled -// day partition. -// -// Step 1 (this file) ships the mechanism WITHOUT touching the serving table: -// a shadow table, lake.signals_latest_daily, is maintained exclusively by the -// daily refresh while the per-pass fold keeps maintaining lake.signals_latest. -// A diff pass after each refresh compares the two over the settled window — -// the production differential evidence that gates the step-4 flip (at which -// point the validated shadow table is promoted and the fold removed). +// History, briefly: the per-pass incremental fold (captureRollupDelta + +// foldSignalsRollup, removed in step 5) kept lake.signals_latest continuously +// exact, but its lake.signals scans carried a correlated bound (s.timestamp >= +// prev_ts) that cannot drive static partition pruning — every span re-opened +// roughly the day partition's current file set, three times, growing all day +// (the daily pass-duration sawtooth). Worse, the fold's DELETE racing din's +// rewrite_data_files compaction silently removed nothing and accumulated +// visible duplicate rows (2026-08-08: 823k rows over 7.7k keys on the live +// table). Since the KV serves signalsLatest (LATEST_KV_READ_MODE=serve), the +// rollup's remaining jobs — KV bootstrap/reconcile source, summaries baseline, +// rare fallback — have no freshness SLA of their own, so it is refreshed once +// daily from a WATERMARK: a constant timestamp literal that prunes the fold to +// the settled day partition. Step 1 (#56) shipped the mechanism against a +// shadow table (lake.signals_latest_daily) with a per-refresh diff as the flip +// evidence — creating that table with plain DDL and never scanning it pre-seed +// (#59), after a zero-row CTAS left degenerate inlined-data state whose first +// scan crashed the ducklake extension (the ducklake#281 family, 2026-08-07); +// the cardinality probe (#62) then quantified the fold-era duplicate +// corruption; step 4 (#63) flipped prod to mode on (promoting the validated +// shadow table, which also remediated that corruption); step 5 removed the +// per-pass fold and the shadow scaffolding. What remains of the shadow era is +// the boot-time promote in LoadDailyRollupState, for a node upgrading straight +// from mode=shadow with a leftover shadow table. // // Invariant (the induction base every fold step relies on): after a refresh to -// watermark W, lake.signals_latest_daily is EXACTLY rollupSelectSQL over -// lake.signals restricted to timestamp < W. Three mechanisms preserve it: +// watermark W, lake.signals_latest is EXACTLY rollupSelectSQL over +// lake.signals restricted to timestamp < W (plus, above W, nothing — the tail +// is the read side's job: the summaries union and the KV serve it). Three +// mechanisms preserve it: // // - the seed: a full bounded recompute (timestamp < W), bucket-chunked; // - the daily fold: rollupSelectSQL over [W_old, W_new) merged onto the @@ -50,7 +59,7 @@ import ( // path records their subjects durably in lake.rollup_late_subjects (same // transaction as the base insert), and the refresh recomputes those // subjects bounded to timestamp < W_new, then clears them. Without this -// the shadow table would be stale for a buffered-upload subject not for +// the rollup would be stale for a buffered-upload subject not for // <=24h but forever. const ( // dailyWatermarkPartition is the lake.ingest_progress key holding the daily @@ -61,8 +70,9 @@ const ( // definition lives in duck (the query layer's summaries union reads it // too); internal/latestkv carries a documented duplicate (import cycle). dailyWatermarkPartition = duck.RollupDailyWatermarkPartition - // dailyRollupTable is the shadow rollup maintained by the daily refresh in - // step 1 (mode shadow). The step-4 flip promotes it to lake.signals_latest. + // dailyRollupTable is the shadow-era rollup table (dq#55 step 1). It is no + // longer written; the name survives only for the boot-time promote/drop of + // a leftover copy in LoadDailyRollupState — after which it is gone forever. dailyRollupTable = "lake.signals_latest_daily" // lateSubjectsTable records subjects that committed base rows stamped // BEFORE the current watermark (late arrivals). Written in the decode @@ -82,48 +92,38 @@ const defaultDailyRollupDelay = 3*time.Hour + 30*time.Minute type DailyRollupMode string const ( - // DailyRollupOff disables the daily refresh entirely (default). + // DailyRollupOff disables the daily refresh entirely. With the per-pass + // fold removed (dq#55 step 5) NOTHING maintains lake.signals_latest in + // this mode — it exists for tests and one-off ops only, and + // LoadDailyRollupState warns when it sees it. DailyRollupOff DailyRollupMode = "off" - // DailyRollupShadow maintains lake.signals_latest_daily by daily refresh - // while the per-pass fold keeps maintaining lake.signals_latest, and diffs - // the two after each refresh — the production differential evidence that - // gates the flip. Serving is untouched. - DailyRollupShadow DailyRollupMode = "shadow" - // DailyRollupOn is the dq#55 step-4 flip: the daily refresh maintains - // lake.signals_latest ITSELF and the per-pass fold is off — span - // transactions lose their dominant, day-length-dependent term. On the - // first boot after switching from shadow, the (validated, one-row-per-key) - // shadow table is PROMOTED into lake.signals_latest — which is also the - // remediation for the duplicate-row corruption the shadow diff exposed - // (2026-08-08: live carried 823k rows over 7.7k keys; the promote - // discards them). Pair with LAKE_ROLLUP_DAILY_SERVING=true on the query - // fleet, or summaries under-count the tail. + // DailyRollupOn (the default): the daily refresh maintains + // lake.signals_latest — since the dq#55 step-4 flip the only writer of the + // rollup, and since step 5 the only mechanism that exists. On the first + // boot after an upgrade straight from mode=shadow, the (validated, + // one-row-per-key) shadow table is PROMOTED into lake.signals_latest — + // which is also the remediation for the duplicate-row corruption the + // shadow diff exposed (2026-08-08: live carried 823k rows over 7.7k keys; + // the promote discards them). Pair with LAKE_ROLLUP_DAILY_SERVING=true on + // the query fleet, or summaries under-count the tail. DailyRollupOn DailyRollupMode = "on" ) -// ParseDailyRollupMode validates a MATERIALIZER_DAILY_ROLLUP_MODE value; empty -// means off. +// ParseDailyRollupMode validates a MATERIALIZER_DAILY_ROLLUP_MODE value. Empty +// means ON: with the per-pass fold gone, an unmaintained rollup must not be +// reachable by default — a node with no explicit mode gets the daily refresh. +// The retired "shadow" value is now invalid so a stale config fails loud at +// boot instead of silently running an unmaintained mode. func ParseDailyRollupMode(s string) (DailyRollupMode, bool) { switch DailyRollupMode(s) { - case "", DailyRollupOff: - return DailyRollupOff, true - case DailyRollupShadow: - return DailyRollupShadow, true - case DailyRollupOn: + case "", DailyRollupOn: return DailyRollupOn, true + case DailyRollupOff: + return DailyRollupOff, true } return DailyRollupOff, false } -// dailyTargetTable is the table the daily refresh maintains: the shadow table -// during the evidence phase, lake.signals_latest itself after the flip. -func (m *DuckLakeMaterializer) dailyTargetTable() string { - if m.dailyMode == DailyRollupOn { - return "lake.signals_latest" - } - return dailyRollupTable -} - // WithDailyRollup configures the daily rollup refresh (dq#55). delay <= 0 uses // defaultDailyRollupDelay. Returns m for chaining. func (m *DuckLakeMaterializer) WithDailyRollup(mode DailyRollupMode, delay time.Duration) *DuckLakeMaterializer { @@ -149,20 +149,19 @@ func (m *DuckLakeMaterializer) dailyActive() bool { // LoadDailyRollupState ensures the daily-refresh tables exist and loads the // watermark. Called once from the Runner before the decode loop (and lazily by -// MaybeDailyRollupRefresh if that call failed); a no-op when the mode is off. +// MaybeDailyRollupRefresh if that call failed); a no-op when the mode is off — +// but a LOUD one: with the fold gone, off means nothing maintains the rollup. func (m *DuckLakeMaterializer) LoadDailyRollupState(ctx context.Context) error { if !m.dailyConfigured() || m.dailyStateLoaded { + if m.dailyMode == DailyRollupOff { + m.log.Warn().Msg("MATERIALIZER_DAILY_ROLLUP_MODE=off: NOTHING maintains lake.signals_latest (the per-pass fold was removed in dq#55 step 5) — tests/one-off ops only") + } return nil } daily, err := m.tableExists(ctx, "lake", "signals_latest_daily") if err != nil { return err } - if !daily && m.dailyMode == DailyRollupShadow { - if err := m.createDailyTable(ctx); err != nil { - return err - } - } if err := m.execRetryConflict(ctx, "CREATE TABLE IF NOT EXISTS "+lateSubjectsTable+" (subject VARCHAR)"); err != nil { return fmt.Errorf("ensuring daily rollup objects: %w", err) } @@ -170,13 +169,14 @@ func (m *DuckLakeMaterializer) LoadDailyRollupState(ctx context.Context) error { if err != nil { return err } - // The shadow→on transition: a leftover shadow table with a valid watermark - // is the validated, one-row-per-key copy — promote it into - // lake.signals_latest (and discard whatever the per-pass fold era left - // there, duplicate-row corruption included). A failure leaves state - // unloaded, so the next caught-up pass retries; the fold is already off - // (mode-gated), and every serving-critical read is KV-backed meanwhile. - if m.dailyMode == DailyRollupOn && daily { + // The shadow→on transition (a node upgrading straight from shadow-era + // config): a leftover shadow table with a valid watermark is the + // validated, one-row-per-key copy — promote it into lake.signals_latest + // (and discard whatever the per-pass fold era left there, duplicate-row + // corruption included). A failure leaves state unloaded, so the next + // caught-up pass retries; every serving-critical read is KV-backed + // meanwhile. After the promote the table is gone forever. + if daily { if w.IsZero() { // A shadow table without a watermark is an aborted shadow seed — // worthless as a promote source. Drop it; the first refresh seeds @@ -252,37 +252,6 @@ func (m *DuckLakeMaterializer) promoteDailyRollup(ctx context.Context) error { return nil } -// createDailyTable creates the shadow table with EXPLICIT DDL — a column-level -// copy of lake.signals_latest's creation statement (setupStatements), and -// deliberately NOT a zero-row CTAS. The original `CREATE ... AS SELECT * FROM -// lake.signals_latest WHERE false` left degenerate inlined-data state on the -// production catalog (Postgres, din's data inlining on): the table's very -// first scan — the seed's bucket-0 DELETE — died inside -// DuckLakeInlinedDataReader::TryInitializeScan with "Attempted to access index -// 0 within vector of size 0" (the ducklake#281 error family, still present at -// v1.5.4), invalidating the embedded database and restart-looping the writer -// (2026-08-07). Every other lake table is created with plain DDL and scans -// fine under inlining; the shadow table now matches. The partition ALTER must -// run only at first creation (re-ALTERing is a crash — see setupStatements). -func (m *DuckLakeMaterializer) createDailyTable(ctx context.Context) error { - stmts := []string{ - `CREATE TABLE IF NOT EXISTS ` + dailyRollupTable + ` ( - subject VARCHAR, subject_bucket INTEGER, name VARCHAR, - "timestamp" TIMESTAMP WITH TIME ZONE, - value_number DOUBLE, value_string VARCHAR, - loc_lat DOUBLE, loc_lon DOUBLE, loc_hdop DOUBLE, loc_heading DOUBLE, - loc_ts TIMESTAMP WITH TIME ZONE, - count BIGINT, first_seen TIMESTAMP WITH TIME ZONE, last_seen TIMESTAMP WITH TIME ZONE)`, - "ALTER TABLE " + dailyRollupTable + " SET PARTITIONED BY (subject_bucket)", - } - for _, s := range stmts { - if err := m.execRetryConflict(ctx, s); err != nil { - return fmt.Errorf("creating daily rollup table: %w", err) - } - } - return nil -} - // execRetryConflict runs one DDL/DML statement, retrying commit conflicts — // the same courtesy ensureSchema extends to din racing catalog maintenance. func (m *DuckLakeMaterializer) execRetryConflict(ctx context.Context, stmt string) error { @@ -401,57 +370,32 @@ func (m *DuckLakeMaterializer) RunDailyRollupRefresh(ctx context.Context, bounda dailyRollupRefreshTotal.WithLabelValues("ok").Inc() m.log.Info().Time("watermark", boundary).Dur("took", time.Since(start)). Msg("daily signals_latest refresh complete") - if m.dailyMode == DailyRollupShadow { - // Diff is evidence, not correctness: a failure must not fail the refresh - // (the watermark has advanced; rerunning the fold would double-fold). - if _, derr := m.DailyRollupDiff(ctx); derr != nil { - m.log.Error().Err(derr).Msg("daily rollup shadow diff failed; no comparison this cycle") - } - } else if err := m.observeRollupCardinality(ctx); err != nil { - // Mode on: the diff (which carried the probe in shadow mode) no longer - // runs, but the one-row-per-key check is the STANDING proof the - // fold-era duplicate corruption stays gone — it must fire every - // refresh, not only at boot. Best-effort like the diff. + // The cardinality probe is evidence, not correctness: a failure must not + // fail the refresh (the watermark has advanced; rerunning the fold would + // double-fold). It fires every refresh — the STANDING proof the fold-era + // duplicate corruption stays gone (dq#64). + if err := m.observeRollupCardinality(ctx); err != nil { m.log.Error().Err(err).Msg("rollup cardinality probe failed; no corruption check this cycle") } return nil } -// seedDailyRollup establishes the induction base: the daily table becomes -// exactly rollupSelectSQL over timestamp < boundary. The table is DROPPED and -// recreated rather than per-bucket DELETEd — the seed must never scan the -// table it is establishing (an empty or half-seeded table's scan is where the -// inlined-reader crash lived, see createDailyTable; a drop is catalog-only), -// and it makes the seed self-healing over ANY damaged prior state, including -// the poisoned CTAS table the first rollout left behind. Then bucket-chunked -// INSERTs like RecomputeRollup (one txn per bucket, memory-bounded over deep -// history). The watermark is written only after every bucket committed, so a -// crash mid-seed simply reseeds from the drop. This is the RecomputeRollup -// cost class, run once at enable (and on operator reseed: delete the -// watermark row). +// seedDailyRollup establishes the induction base: lake.signals_latest becomes +// exactly rollupSelectSQL over timestamp < boundary. The table is being SERVED +// — never drop it. It is cleared transactionally instead: readers see the old +// content until the commit, then a briefly-empty rollup that fills bucket by +// bucket — the same partial visibility the LAKE_REBUILD_ROLLUP_ON_BOOT +// recovery has always had, and every serving-critical read is KV-backed +// anyway. Then bucket-chunked INSERTs like RecomputeRollup (one txn per +// bucket, memory-bounded over deep history). The watermark is written only +// after every bucket committed, so a crash mid-seed simply reseeds from the +// clear. This is the RecomputeRollup cost class, run once at enable (and on +// operator reseed: delete the watermark row). func (m *DuckLakeMaterializer) seedDailyRollup(ctx context.Context, boundary time.Time) error { - target := m.dailyTargetTable() - m.log.Info().Time("boundary", boundary).Str("table", target). + m.log.Info().Time("boundary", boundary). Msg("seeding the daily rollup (bounded full recompute; one-time, O(history))") - if target == dailyRollupTable { - // Shadow target: nothing reads it, so DROP+recreate (never scan a table - // being established — see createDailyTable's crash history). - if err := m.execRetryConflict(ctx, "DROP TABLE IF EXISTS "+dailyRollupTable); err != nil { - return fmt.Errorf("daily seed drop: %w", err) - } - if err := m.createDailyTable(ctx); err != nil { - return err - } - } else { - // Live target (mode on, fresh install or operator reseed): the table is - // being SERVED — never drop it. Clear it transactionally instead; - // readers see the old content until the commit, then a briefly-empty - // rollup that fills bucket by bucket — the same partial visibility the - // LAKE_REBUILD_ROLLUP_ON_BOOT recovery has always had, and every - // serving-critical read is KV-backed anyway. - if err := m.execRetryConflict(ctx, "DELETE FROM "+target); err != nil { - return fmt.Errorf("daily seed clear: %w", err) - } + if err := m.execRetryConflict(ctx, "DELETE FROM lake.signals_latest"); err != nil { + return fmt.Errorf("daily seed clear: %w", err) } bound := fmt.Sprintf(`"timestamp" < make_timestamp(%d)`, boundary.UnixMicro()) for b := 0; b < duck.NumLatestBuckets; b++ { @@ -463,7 +407,7 @@ func (m *DuckLakeMaterializer) seedDailyRollup(ctx context.Context, boundary tim if err != nil { return err } - if _, err := tx.ExecContext(ctx, "INSERT INTO "+target+signalsLatestColumns+rollupSelectSQL(where)); err != nil { + if _, err := tx.ExecContext(ctx, "INSERT INTO lake.signals_latest"+signalsLatestColumns+rollupSelectSQL(where)); err != nil { _ = tx.Rollback() return fmt.Errorf("daily seed bucket %d insert: %w", b, err) } @@ -493,14 +437,14 @@ func (m *DuckLakeMaterializer) seedDailyRollup(ctx context.Context, boundary tim return tx.Commit() } -// foldDailyRollup folds [from, to) into the daily table and advances the +// foldDailyRollup folds [from, to) into lake.signals_latest and advances the // watermark in the same transaction, then recomputes the late set. The tail // aggregate IS rollupSelectSQL restricted to the window — a constant-literal // predicate on the partition column's source, so the scan prunes to the // settled day partition(s) instead of re-deriving bounds per row (the whole // point of dq#55). func (m *DuckLakeMaterializer) foldDailyRollup(ctx context.Context, from, to time.Time) error { - target := m.dailyTargetTable() + const target = "lake.signals_latest" window := fmt.Sprintf(`WHERE "timestamp" >= make_timestamp(%d) AND "timestamp" < make_timestamp(%d)`, from.UnixMicro(), to.UnixMicro()) tx, err := m.db.BeginTx(ctx, nil) @@ -599,7 +543,7 @@ func (m *DuckLakeMaterializer) recomputeLateDailySubjects(ctx context.Context, b } if len(subjects) > m.maxDirtySubjects { m.log.Warn().Int("late_subjects", len(subjects)). - Msg("daily rollup late set overflowed; reseeding the daily table instead of per-subject recompute") + Msg("daily rollup late set overflowed; reseeding lake.signals_latest instead of per-subject recompute") if err := m.execRetryConflict(ctx, "DELETE FROM "+lateSubjectsTable); err != nil { return err } @@ -636,9 +580,9 @@ func (m *DuckLakeMaterializer) recomputeLateDailySubjects(ctx context.Context, b return nil } -// recomputeDailyChunk DELETEs+recomputes one bucket's given subjects in the -// daily table (bounded by `bound`) and clears their late marks, all in one -// transaction — the marks disappear exactly when the recompute that makes +// recomputeDailyChunk DELETEs+recomputes one bucket's given subjects in +// lake.signals_latest (bounded by `bound`) and clears their late marks, all in +// one transaction — the marks disappear exactly when the recompute that makes // them unnecessary lands. func (m *DuckLakeMaterializer) recomputeDailyChunk(ctx context.Context, bucket int, subjects []string, bound string) error { args := make([]any, len(subjects)) @@ -655,10 +599,10 @@ func (m *DuckLakeMaterializer) recomputeDailyChunk(ctx context.Context, bucket i } defer func() { _ = tx.Rollback() }() if _, err := tx.ExecContext(ctx, - fmt.Sprintf("DELETE FROM %s WHERE subject_bucket = %d AND subject IN (%s)", m.dailyTargetTable(), bucket, in), args...); err != nil { + fmt.Sprintf("DELETE FROM lake.signals_latest WHERE subject_bucket = %d AND subject IN (%s)", bucket, in), args...); err != nil { return fmt.Errorf("delete: %w", err) } - if _, err := tx.ExecContext(ctx, "INSERT INTO "+m.dailyTargetTable()+signalsLatestColumns+rollupSelectSQL(where), args...); err != nil { + if _, err := tx.ExecContext(ctx, "INSERT INTO lake.signals_latest"+signalsLatestColumns+rollupSelectSQL(where), args...); err != nil { return fmt.Errorf("insert: %w", err) } if _, err := tx.ExecContext(ctx, @@ -759,107 +703,34 @@ func (m *DuckLakeMaterializer) PersistDailyLateSubjects(ctx context.Context) err return nil } -// DailyRollupDiff compares the daily table against the live rollup over the -// settled window (live rows with last_seen < watermark; fresher live rows are -// legitimately ahead of the daily table and excluded). This is the -// production differential evidence for the dq#55 flip: classes are -// missing_daily (live has a settled row the daily table lacks), missing_live -// (the daily table has a row live lacks — also what retention-prune drift -// looks like), and mismatch (any column differs). Zero across days of real -// traffic is the gate. -func (m *DuckLakeMaterializer) DailyRollupDiff(ctx context.Context) (map[string]int, error) { - start := time.Now() - stmt := fmt.Sprintf(` -SELECT - CASE WHEN l.subject IS NULL THEN 'missing_live' - WHEN d.subject IS NULL THEN 'missing_daily' - ELSE 'mismatch' END AS class, - coalesce(l.subject, d.subject) AS subject, coalesce(l.name, d.name) AS name -FROM lake.signals_latest l -FULL OUTER JOIN %s d ON d.subject = l.subject AND d.name = l.name -WHERE (l.subject IS NULL OR l.last_seen < make_timestamp(%d)) - AND (l.subject IS NULL OR d.subject IS NULL - OR l.count != d.count - OR l."timestamp" != d."timestamp" - OR l.value_number IS DISTINCT FROM d.value_number - OR l.value_string IS DISTINCT FROM d.value_string - OR l.loc_lat != d.loc_lat OR l.loc_lon != d.loc_lon - OR l.loc_hdop != d.loc_hdop OR l.loc_heading != d.loc_heading - OR coalesce(l.loc_ts, make_timestamp(0)) != coalesce(d.loc_ts, make_timestamp(0)) - OR l.first_seen != d.first_seen OR l.last_seen != d.last_seen)`, - dailyRollupTable, m.dailyWatermark.UnixMicro()) - rows, err := m.db.QueryContext(ctx, stmt) - if err != nil { - return nil, fmt.Errorf("daily rollup diff: %w", err) - } - defer rows.Close() //nolint:errcheck - counts := map[string]int{} - logged := 0 - for rows.Next() { - var class, subject, name string - if err := rows.Scan(&class, &subject, &name); err != nil { - return nil, fmt.Errorf("scanning diff row: %w", err) - } - counts[class]++ - if logged < 10 { - logged++ - m.log.Warn().Str("class", class).Str("subject", subject).Str("name", name). - Msg("daily rollup shadow diff: daily table disagrees with the incremental rollup") - } - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("diff scan: %w", err) - } - for _, class := range []string{"missing_daily", "missing_live", "mismatch"} { - dailyRollupDiffRows.WithLabelValues(class).Set(float64(counts[class])) - } - total := counts["missing_daily"] + counts["missing_live"] + counts["mismatch"] - m.log.Info().Int("diff_rows", total).Dur("took", time.Since(start)). - Msg("daily rollup shadow diff complete") - if err := m.observeRollupCardinality(ctx); err != nil { - m.log.Error().Err(err).Msg("rollup cardinality probe failed") - } - return counts, nil -} - -// observeRollupCardinality measures, for each rollup table, how many physical +// observeRollupCardinality measures how many physical lake.signals_latest // rows exist per (subject, name) key — the rollup contract is EXACTLY ONE. // Added when the 2026-08-08 shadow diff hit 300k all-mismatch rows over a -// ≤39k-key fleet: a FULL OUTER JOIN can only exceed the key count if a side -// holds duplicate keys, and this probe is what says which side and how badly. -// The standing suspicion is the per-pass fold's DELETE racing din's -// rewrite_data_files compaction (the delete lands on rows whose files were -// just rewritten and removes nothing; din's 2026-08-06 flush conflict is the -// same collision seen from the other side). Duplicate keys on the LIVE table -// mean dataSummary/rollup-fallback reads are serving duplicated rows; the -// dq#55 flip (one fold/day, promote the clean daily table) is the structural -// fix, and this gauge is how we prove the corruption gone afterwards. +// ≤39k-key fleet: the cause was the per-pass fold's DELETE racing din's +// rewrite_data_files compaction (the delete landed on rows whose files were +// just rewritten and removed nothing; din's 2026-08-06 flush conflict is the +// same collision seen from the other side). Duplicate keys mean +// dataSummary/rollup-fallback reads serve duplicated rows; the daily refresh +// (one fold/day, no per-pass DELETE) is the structural fix, and this gauge is +// the standing proof the corruption stays gone. func (m *DuckLakeMaterializer) observeRollupCardinality(ctx context.Context) error { - sides := []struct{ label, table string }{{"live", "lake.signals_latest"}} - if m.dailyMode == DailyRollupShadow { - // The shadow table exists only during the evidence phase; post-flip the - // live table IS the daily-maintained one. - sides = append(sides, struct{ label, table string }{"daily", dailyRollupTable}) - } - for _, side := range sides { - var keys, rows, dupKeys, dupRows int64 - q := fmt.Sprintf(`SELECT count(*), coalesce(sum(n), 0), - count(*) FILTER (WHERE n > 1), coalesce(sum(n) FILTER (WHERE n > 1), 0) - FROM (SELECT count(*) AS n FROM %s GROUP BY subject, name)`, side.table) - if err := m.db.QueryRowContext(ctx, q).Scan(&keys, &rows, &dupKeys, &dupRows); err != nil { - return fmt.Errorf("cardinality of %s: %w", side.table, err) - } - dailyRollupSideRows.WithLabelValues(side.label, "keys").Set(float64(keys)) - dailyRollupSideRows.WithLabelValues(side.label, "rows").Set(float64(rows)) - dailyRollupSideRows.WithLabelValues(side.label, "dup_keys").Set(float64(dupKeys)) - dailyRollupSideRows.WithLabelValues(side.label, "dup_rows").Set(float64(dupRows)) - evt := m.log.Info() - if dupKeys > 0 { - evt = m.log.Warn() - } - evt.Str("table", side.table).Int64("keys", keys).Int64("rows", rows). - Int64("dup_keys", dupKeys).Int64("dup_rows", dupRows). - Msg("rollup cardinality (rows must equal keys; every excess row is a visible duplicate)") - } + var keys, rows, dupKeys, dupRows int64 + if err := m.db.QueryRowContext(ctx, `SELECT count(*), coalesce(sum(n), 0), + count(*) FILTER (WHERE n > 1), coalesce(sum(n) FILTER (WHERE n > 1), 0) + FROM (SELECT count(*) AS n FROM lake.signals_latest GROUP BY subject, name)`). + Scan(&keys, &rows, &dupKeys, &dupRows); err != nil { + return fmt.Errorf("cardinality of lake.signals_latest: %w", err) + } + dailyRollupSideRows.WithLabelValues("live", "keys").Set(float64(keys)) + dailyRollupSideRows.WithLabelValues("live", "rows").Set(float64(rows)) + dailyRollupSideRows.WithLabelValues("live", "dup_keys").Set(float64(dupKeys)) + dailyRollupSideRows.WithLabelValues("live", "dup_rows").Set(float64(dupRows)) + evt := m.log.Info() + if dupKeys > 0 { + evt = m.log.Warn() + } + evt.Int64("keys", keys).Int64("rows", rows). + Int64("dup_keys", dupKeys).Int64("dup_rows", dupRows). + Msg("rollup cardinality (rows must equal keys; every excess row is a visible duplicate)") return nil } diff --git a/internal/materializer/daily_rollup_internal_test.go b/internal/materializer/daily_rollup_internal_test.go index ca145d8..fb74e2f 100644 --- a/internal/materializer/daily_rollup_internal_test.go +++ b/internal/materializer/daily_rollup_internal_test.go @@ -49,10 +49,10 @@ func TestParseDailyRollupMode(t *testing.T) { want DailyRollupMode ok bool }{ - {"", DailyRollupOff, true}, + {"", DailyRollupOn, true}, // empty defaults ON: an unmaintained rollup must not be reachable by default {"off", DailyRollupOff, true}, - {"shadow", DailyRollupShadow, true}, - {"on", DailyRollupOn, true}, // the step-4 flip + {"on", DailyRollupOn, true}, + {"shadow", DailyRollupOff, false}, // retired in step 5: stale shadow-era configs must fail loud {"bogus", DailyRollupOff, false}, } { got, ok := ParseDailyRollupMode(tc.in) diff --git a/internal/materializer/ducklake.go b/internal/materializer/ducklake.go index b478800..6eeb3c7 100644 --- a/internal/materializer/ducklake.go +++ b/internal/materializer/ducklake.go @@ -1556,38 +1556,21 @@ func (m *DuckLakeMaterializer) insertDecodedSteady(ctx context.Context, tx *sql. } cleanup = append(cleanup, tmp) tsMin, tsMax := timeRange(dec.signals, func(r SignalRow) time.Time { return r.Timestamp }) - // #5b: steady-state lake.signals_latest is maintained INCREMENTALLY here - // (O(batch), not an O(history) recompute per flush). The count delta must be - // captured BEFORE the base insert — afterwards the batch rows are in the base and - // the NOT-EXISTS probe finds them, yielding delta 0 (that is exactly what makes a - // replayed window idempotent). Backfill (bulk/arbitrarily-old) skips the fold and - // defers to the end-of-catch-up recompute (markDirtyFromBatch marks it dirty). - // - // Under MATERIALIZER_DAILY_ROLLUP_MODE=on (dq#55 step 4) the fold is OFF: - // the daily refresh maintains lake.signals_latest, and these three - // lake.signals scans — the dominant, day-length-dependent term in span - // cost — leave the span transaction entirely. This is the flip #55 exists - // for; the fold code itself is removed in step 5. - foldOn := !m.backfillMode && m.dailyMode != DailyRollupOn - if foldOn { - if err := m.captureRollupDelta(ctx, tx, tmp); err != nil { - return cleanup, err - } - } + // lake.signals_latest is NOT touched here (dq#55 step 5): the per-pass + // incremental fold (#5b) was removed once the daily watermarked refresh + // (daily_rollup.go) became the rollup's only writer — its three + // lake.signals scans were the dominant, day-length-dependent term in + // span cost, and its DELETE raced din's compaction into duplicate rows. + // The span transaction is now base-insert + late-subject marking only. if _, err := tx.ExecContext(ctx, antiJoinInsert("lake.signals", tmp, tsMin, tsMax, m.backfillMode)); err != nil { return cleanup, fmt.Errorf("insert signals: %w", err) } - if foldOn { - if err := m.foldSignalsRollup(ctx, tx, tmp); err != nil { - return cleanup, err - } - } // Record late arrivals for the daily rollup refresh (dq#55): a row // stamped before the daily watermark is invisible to every future // constant-predicate fold, so its subject must be marked for the // refresh's bounded recompute. In THIS transaction, so the mark is // crash-atomic with the rows it marks. Normally zero rows and skipped - // entirely; a no-op unless MATERIALIZER_DAILY_ROLLUP_MODE is set. + // entirely; a no-op until the daily watermark is loaded. if m.dailyActive() { if stmt, args := m.lateDailyInsert(dec.signals); stmt != "" { if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { @@ -1613,9 +1596,11 @@ func (m *DuckLakeMaterializer) insertDecodedSteady(ctx context.Context, tx *sql. // markDirtyFromBatch records the subjects dec wrote so the decoupled rollups refresh // only their rows, escalating to a full rebuild if a dirty set overflows its cap (a // fleet-wide catch-up would otherwise grow the maps unbounded — see the field docs). -// signals_latest is now maintained incrementally at commit time (#5b), so signals are -// dirtied ONLY under backfillMode (the deferred bulk catch-up recomputes them at the -// end); events_latest still uses the decoupled recompute, so events are always dirtied. +// signals_latest is maintained by the daily refresh (dq#55), so signals are dirtied +// ONLY under backfillMode — the bulk catch-up's touched set, which +// PersistDailyLateSubjects hands to the late table (and FlushRollup recomputes +// directly in the tests-only mode off); events_latest still uses the decoupled +// recompute, so events are always dirtied. // Single-writer: mutated only on the decode-loop goroutine. func (m *DuckLakeMaterializer) markDirtyFromBatch(dec *decodedBatch) { if m.backfillMode { @@ -1642,103 +1627,6 @@ func (m *DuckLakeMaterializer) markDirtyFromBatch(dec *decodedBatch) { } } -// captureRollupDelta stages, into the per-connection temp table _rollup_delta, the number -// of NEWLY-DISTINCT (subject, name, timestamp) tuples this batch adds — i.e. the exact -// increment to lake.signals_latest.count (#5b). It must run BEFORE the base insert: it -// probes lake.signals for tuples that DON'T already exist, so a redelivery or a -// same-(subject,name,timestamp) collision (which the count must not double) contributes 0, -// and a replayed window (rows already at rest) yields 0 — the idempotency the crash- -// recovery path relies on. The probe matches on the EXACT timestamp (partition-pruned by -// subject_bucket + the day partition), deliberately NOT clamped to the anti-join's 30d -// dedup window: an ancient (>30d) redelivery must still find its existing row so count -// stays equal to a full RecomputeRollup, even though the physical anti-join may re-insert a -// duplicate row (the read-path QUALIFY dedup collapses that, and count must match it). -func (m *DuckLakeMaterializer) captureRollupDelta(ctx context.Context, tx *sql.Tx, sigParquet string) error { - q := fmt.Sprintf(`CREATE OR REPLACE TEMPORARY TABLE _rollup_delta AS -SELECT b.subject, b.name, CAST(count(*) AS BIGINT) AS delta -FROM (SELECT DISTINCT subject, name, subject_bucket, "timestamp" FROM read_parquet(%[1]s)) b -WHERE NOT EXISTS ( - SELECT 1 FROM lake.signals s - WHERE s.subject_bucket = b.subject_bucket AND s.subject = b.subject AND s.name = b.name - AND s."timestamp" = b."timestamp" -) -GROUP BY b.subject, b.name`, sqlLit(sigParquet)) - if _, err := tx.ExecContext(ctx, q); err != nil { - return fmt.Errorf("capture rollup count delta: %w", err) - } - return nil -} - -// foldSignalsRollup folds this batch into lake.signals_latest incrementally, EXACTLY as a -// full recompute would (proven by the differential test), but O(batch) instead of -// O(history) (#5b). It runs AFTER the base insert, inside the same transaction: -// - RECENCY (timestamp, value_*, loc_*, loc_ts) is recomputed from the base but BOUNDED -// to "timestamp >= the row's prior latest" — the new latest is either in this batch -// (newer) or unchanged (the prior row still qualifies), so the bound is exact yet -// prunes every day-partition older than the prior latest. This makes recency -// SELF-HEALING (recomputed from the base each batch), matching the recompute's deduped -// arg_max via ORDER BY timestamp DESC, cloud_event_id ASC. -// - COUNT is prev.count + the captured NOT-EXISTS delta; FIRST_SEEN is min(prev, batch). -// These carry forward (idempotent on replay), and self-heal via the boot rebuild -// (RecomputeRollup / LAKE_REBUILD_ROLLUP_ON_BOOT) if a rollup row is ever lost. -// A (subject,name) with no prior rollup row folds against zero — correct in steady state -// (a newly-seen signal has no prior base); a mass-loss (dropped rollup) is the boot -// rebuild's job, exactly as before. -func (m *DuckLakeMaterializer) foldSignalsRollup(ctx context.Context, tx *sql.Tx, sigParquet string) error { - build := fmt.Sprintf(`CREATE OR REPLACE TEMPORARY TABLE _rollup_new AS -WITH affected AS ( - SELECT subject, name, any_value(subject_bucket) AS subject_bucket, min("timestamp") AS batch_min - FROM read_parquet(%[1]s) GROUP BY subject, name -), -prev AS ( - SELECT l.subject, l.name, l."timestamp" AS prev_ts, l.loc_ts AS prev_loc_ts, l.count AS prev_count, l.first_seen AS prev_first - FROM lake.signals_latest l - WHERE EXISTS (SELECT 1 FROM affected a WHERE a.subject = l.subject AND a.name = l.name) -), -recency AS ( - SELECT s.subject, s.name, s."timestamp" AS ts, s.value_number, s.value_string - FROM lake.signals s - JOIN affected a ON s.subject = a.subject AND s.name = a.name AND s.subject_bucket = a.subject_bucket - LEFT JOIN prev p ON p.subject = s.subject AND p.name = s.name - WHERE s."timestamp" >= coalesce(p.prev_ts, make_timestamp(0)) - QUALIFY row_number() OVER (PARTITION BY s.subject, s.name ORDER BY s."timestamp" DESC, s.cloud_event_id ASC) = 1 -), -locrec AS ( - SELECT s.subject, s.name, s."timestamp" AS loc_ts, s.loc_lat, s.loc_lon, s.loc_hdop, s.loc_heading - FROM lake.signals s - JOIN affected a ON s.subject = a.subject AND s.name = a.name AND s.subject_bucket = a.subject_bucket - LEFT JOIN prev p ON p.subject = s.subject AND p.name = s.name - WHERE (s.loc_lat != 0 OR s.loc_lon != 0) AND s."timestamp" >= coalesce(p.prev_loc_ts, make_timestamp(0)) - QUALIFY row_number() OVER (PARTITION BY s.subject, s.name ORDER BY s."timestamp" DESC, s.cloud_event_id ASC) = 1 -) -SELECT a.subject, a.subject_bucket, a.name, - r.ts AS "timestamp", r.value_number, r.value_string, - coalesce(lr.loc_lat, 0) AS loc_lat, coalesce(lr.loc_lon, 0) AS loc_lon, - coalesce(lr.loc_hdop, 0) AS loc_hdop, coalesce(lr.loc_heading, 0) AS loc_heading, - coalesce(lr.loc_ts, make_timestamp(0)) AS loc_ts, - coalesce(p.prev_count, 0) + coalesce(d.delta, 0) AS count, - LEAST(coalesce(p.prev_first, a.batch_min), a.batch_min) AS first_seen, - r.ts AS last_seen -FROM affected a -JOIN recency r ON r.subject = a.subject AND r.name = a.name -LEFT JOIN locrec lr ON lr.subject = a.subject AND lr.name = a.name -LEFT JOIN prev p ON p.subject = a.subject AND p.name = a.name -LEFT JOIN _rollup_delta d ON d.subject = a.subject AND d.name = a.name`, sqlLit(sigParquet)) - if _, err := tx.ExecContext(ctx, build); err != nil { - return fmt.Errorf("build incremental rollup rows: %w", err) - } - if _, err := tx.ExecContext(ctx, - `DELETE FROM lake.signals_latest WHERE EXISTS (SELECT 1 FROM _rollup_new n WHERE n.subject = lake.signals_latest.subject AND n.name = lake.signals_latest.name)`); err != nil { - return fmt.Errorf("delete superseded rollup rows: %w", err) - } - if _, err := tx.ExecContext(ctx, - `INSERT INTO lake.signals_latest (subject, subject_bucket, name, "timestamp", value_number, value_string, loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen) - SELECT subject, subject_bucket, name, "timestamp", value_number, value_string, loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen FROM _rollup_new`); err != nil { - return fmt.Errorf("insert incremental rollup rows: %w", err) - } - return nil -} - // writeWindow writes an INTERMEDIATE pagination window (finding #1c): the decoded rows // go into lake.signals/events in their own transaction that does NOT advance the ingest // cursor. It is idempotent (the cloud_event_id anti-join) and cursor-independent, so — @@ -1981,23 +1869,6 @@ func (m *DuckLakeMaterializer) PruneDecoded(ctx context.Context, retention time. WHERE s.subject_bucket = sl.subject_bucket AND s.subject = sl.subject AND s.name = sl.name)`, cutoff)); err != nil { return total, fmt.Errorf("pruning orphaned rollup rows: %w", err) } - // Same orphan cleanup for the daily rollup shadow table (dq#55): without it a - // retention prune strips live rollup rows the daily table still carries, and - // the shadow diff reads that drift as permanent missing_live noise. Gated on - // a NON-ZERO watermark, not just loaded state: pre-seed the table is empty - // (or damaged from an aborted seed) and must not be scanned — the seed - // itself never scans it for the same reason (see seedDailyRollup). Shadow - // mode only: post-flip there is no second table (the live block above - // covers the daily-maintained lake.signals_latest). - if m.dailyMode == DailyRollupShadow && m.dailyStateLoaded && !m.dailyWatermark.IsZero() { - if _, err := m.db.ExecContext(ctx, - fmt.Sprintf(`DELETE FROM %s sd WHERE sd.last_seen < make_timestamp(%d) AND NOT EXISTS ( - SELECT 1 FROM lake.signals s - WHERE s.subject_bucket = sd.subject_bucket AND s.subject = sd.subject AND s.name = sd.name)`, - dailyRollupTable, cutoff)); err != nil { - return total, fmt.Errorf("pruning orphaned daily rollup rows: %w", err) - } - } // Same orphan cleanup for the events rollup (finding #5a): drop events_latest rows // whose base events were all pruned away, bounded to last_seen < cutoff so the // anti-join probes only long-dormant rows and each NOT EXISTS is partition-pruned. @@ -2025,25 +1896,14 @@ func (m *DuckLakeMaterializer) PruneDecoded(ctx context.Context, retention time. const signalsLatestColumns = ` (subject, subject_bucket, name, "timestamp", value_number, ` + `value_string, loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen) ` -// rollupSelectSQL is the FULL-history recompute of a set of rollup rows. Steady-state -// maintenance no longer uses it — lake.signals_latest is folded INCREMENTALLY at commit -// time (foldSignalsRollup, #5b), O(batch) not O(history). This recompute is retained for -// the paths that genuinely need a from-scratch rebuild: the disaster-recovery / boot -// rebuild (RecomputeRollup / LAKE_REBUILD_ROLLUP_ON_BOOT) and the deferred bulk-backfill -// catch-up (FlushRollup over backfill-dirtied subjects). Kept byte-identical to the fold's -// result — the differential test (tests/ducklake_incremental_rollup_test.go) asserts the -// incremental path equals this recompute across redelivery, same-timestamp collision, -// out-of-order arrival, multi-window spans, and crash-replay. -// -// Why the fold is exact where a naive `timestamp >= floor` recompute would not be: -// RECENCY (timestamp/value_*/loc_*/loc_ts) IS recomputed each batch, but bounded to -// `timestamp >= the row's prior latest` — the new latest is either in the batch or the -// prior row still qualifies, so the bound prunes old day-partitions without dropping the -// answer (self-healing + exact). COUNT and FIRST_SEEN are the full-history aggregates a -// floor would corrupt, so they are NOT floored: count carries forward as prev.count + a -// NOT-EXISTS delta over DISTINCT (subject,name,timestamp) — collisions and redeliveries -// contribute 0, and a replayed window contributes 0 (idempotent) — and first_seen min-folds -// the batch min. Both self-heal via this recompute on boot if a rollup row is ever lost. +// rollupSelectSQL is the FULL-history recompute of a set of rollup rows — the single +// definition of what a lake.signals_latest row IS. Every writer derives from it: the +// daily refresh (dq#55) seeds with it bounded to timestamp < W, folds with it over +// [W_old, W_new), and recomputes late subjects with it bounded to < W; the +// disaster-recovery / boot rebuild (RecomputeRollup / LAKE_REBUILD_ROLLUP_ON_BOOT) +// and the tests-only mode-off FlushRollup run it unbounded. (The per-pass +// incremental fold that once maintained the rollup at commit time (#5b) was removed +// in dq#55 step 5 — the daily refresh is the only steady-state writer.) func rollupSelectSQL(whereClause string) string { const locNonzero = "(loc_lat != 0 OR loc_lon != 0)" return fmt.Sprintf(`SELECT subject, any_value(subject_bucket) AS subject_bucket, name, @@ -2069,8 +1929,11 @@ func rollupSelectSQL(whereClause string) string { const rollupSubjectChunk = 500 // FlushRollup recomputes lake.signals_latest for every subject whose base rows -// changed since the last flush (the decoupled, off-commit rollup maintenance). -// A no-op when nothing is dirty. Subject-scoped, not bucket-scoped (B2): a +// changed since the last flush. Since dq#55 step 5 the signals path below the +// mode gate runs only under the tests-only mode off (steady state dirties +// signal subjects solely in backfillMode, and mode on diverts those to the +// late set); it is kept because mode-off tests and the backfill flush still +// exercise it. A no-op when nothing is dirty. Subject-scoped, not bucket-scoped (B2): a // bucket recompute is O(the bucket's entire retained history) and bucket // dirtiness saturates at trivial fleet activity, which made every flush a // full-table recompute on the decode goroutine. Recomputing only the dirty diff --git a/internal/materializer/materializer.go b/internal/materializer/materializer.go index 4564ecc..c3a78f7 100644 --- a/internal/materializer/materializer.go +++ b/internal/materializer/materializer.go @@ -161,11 +161,12 @@ func (r *Runner) Run(ctx context.Context) error { failures.record(true, time.Now()) // any success clears the failure streak triedSessionRecycle = false // a healthy pass proves the session pool recovered if processed > 0 { - // Still draining. signals_latest is maintained INCREMENTALLY at commit - // (#5b), so FlushRollup here only recomputes the events_latest rollup and - // any backfill-dirtied signal subjects — cheap in steady state. Interval- - // gated so a long catch-up doesn't stall the drain (backfill defers to the - // single catch-up flush so the drain runs flat-out). + // Still draining. signals_latest is maintained by the daily refresh + // (dq#55), so FlushRollup here only recomputes the events_latest rollup + // (signal subjects are dirtied solely in backfill mode) — cheap in steady + // state. Interval-gated so a long catch-up doesn't stall the drain + // (backfill defers to the single catch-up flush so the drain runs + // flat-out). if !r.cfg.BackfillMode { r.maybeFlushRollup(ctx, &lastRollup) } @@ -184,8 +185,8 @@ func (r *Runner) Run(ctx context.Context) error { // PollInterval (15s) regardless of RollupInterval — churning ~256 tiny // files per flush into the 256-way-partitioned rollup tables. Honoring // RollupInterval here makes MATERIALIZER_ROLLUP_INTERVAL actually govern - // the steady-state cadence (signals_latest stays fresh either way — it is - // folded incrementally at commit, so this flush is a no-op for it). + // the steady-state cadence (signals_latest is the daily refresh's job, so + // this flush is a no-op for it). r.maybeFlushRollup(ctx, &lastRollup) // The daily rollup refresh (dq#55) runs only from the caught-up // branch: "caught up" is the settled-cursor condition its boundary diff --git a/internal/materializer/metrics.go b/internal/materializer/metrics.go index b5d31f8..212a39a 100644 --- a/internal/materializer/metrics.go +++ b/internal/materializer/metrics.go @@ -152,7 +152,7 @@ var ( // resolve_blobs — S3 GETs for externalized payloads (16-way concurrent) // decode — model-garage conversion fan-out (CPU) // write_window — one intermediate window's DuckLake txn (temp parquet, - // anti-join INSERT, incremental rollup fold, KV publish) + // anti-join INSERT, late-subject marking, KV publish) // commit — the final window's txn incl. the cursor CAS phaseSeconds = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "dq_materializer_phase_seconds", @@ -161,8 +161,7 @@ var ( }, []string{"phase"}) // Daily rollup refresh metrics (dq#55; daily_rollup.go). The watermark gauge // is the liveness signal: it must advance once per UTC day — alert on it - // falling more than ~2 days behind now. The diff gauges are the shadow-mode - // differential evidence; any sustained non-zero blocks the step-4 flip. + // falling more than ~2 days behind now. dailyRollupRefreshSeconds = prometheus.NewGauge(prometheus.GaugeOpts{ Name: "dq_materializer_daily_rollup_refresh_seconds", Help: "Wall-clock of the most recent daily signals_latest refresh (seed or fold + late-set recompute).", @@ -175,10 +174,6 @@ var ( Name: "dq_materializer_daily_rollup_watermark_timestamp_seconds", Help: "The daily rollup watermark (UTC-midnight boundary the daily table is exact through), as a unix timestamp. Advances once per day; alert if it falls >2 days behind.", }) - dailyRollupDiffRows = prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Name: "dq_materializer_daily_rollup_diff_rows", - Help: "Shadow-mode diff between the daily table and the incremental rollup over the settled window, by class (missing_daily|missing_live|mismatch). Must hold at zero to gate the dq#55 flip.", - }, []string{"class"}) dailyRollupLateSubjects = prometheus.NewGauge(prometheus.GaugeOpts{ Name: "dq_materializer_daily_rollup_late_subjects", Help: "Late-arrival subjects (rows stamped before the watermark) recomputed by the most recent daily refresh.", @@ -188,7 +183,7 @@ var ( // serving-visible duplicate rows — see observeRollupCardinality). dailyRollupSideRows = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "dq_materializer_daily_rollup_table_rows", - Help: "Rollup table cardinality by side (live|daily) and kind (keys|rows|dup_keys|dup_rows), measured at each daily refresh. rows > keys means visible duplicate rows.", + Help: "lake.signals_latest cardinality by kind (keys|rows|dup_keys|dup_rows), measured at boot and at each daily refresh; side is always \"live\" (the shadow side retired with dq#55 step 5). rows > keys means visible duplicate rows.", }, []string{"side", "kind"}) // progressReportErrorsTotal counts failures writing dq's snapshot floor to // meta.din_consumer_progress. Decode keeps succeeding (a separate txn) so dq's own @@ -218,7 +213,7 @@ func registerMetrics() { headSnapshotID, cursorResetGap, blobMissingTotal, blobPoisonTotal, phaseSeconds, progressReportErrorsTotal, dailyRollupRefreshSeconds, dailyRollupRefreshTotal, dailyRollupWatermark, - dailyRollupDiffRows, dailyRollupLateSubjects, dailyRollupSideRows, + dailyRollupLateSubjects, dailyRollupSideRows, ) }) } diff --git a/internal/service/duck/kv_latest.go b/internal/service/duck/kv_latest.go index b2eb8bd..647a587 100644 --- a/internal/service/duck/kv_latest.go +++ b/internal/service/duck/kv_latest.go @@ -26,7 +26,9 @@ const ( // KVReadShadow serves from the rollup exactly as before but ALSO reads the // cache and compares, counting dq_lake_latest_kv_shadow_total — the dark // launch that proves per-query parity on real traffic before any user - // request depends on the cache. + // request depends on the cache. Under the daily rollup (dq#55) the rollup + // baseline is day-stale, so shadow comparisons mint kv_newer noise — shadow + // remains useful only for gross-mismatch detection. KVReadShadow KVReadMode = "shadow" // KVReadServe answers from the cache, falling back to the rollup path on // any miss, error, or unknown entry version. NATS unavailability degrades diff --git a/internal/service/duck/metrics.go b/internal/service/duck/metrics.go index 62558e0..b5f9ad1 100644 --- a/internal/service/duck/metrics.go +++ b/internal/service/duck/metrics.go @@ -200,7 +200,9 @@ var kvShadowTotal = promauto.NewCounterVec(prometheus.CounterOpts{ // kvExtShadowTotal is kvShadowTotal for the extended KV paths (dq#55 step 3: // allLatest, availableSignals), split by query so each move gates its own // serve flip. mismatch must stay at zero before LATEST_KV_READ_MODE_EXTENDED -// flips to serve. +// flips to serve. Under the daily rollup (dq#55) the rollup baseline is +// day-stale, so shadow comparisons mint kv_newer noise — shadow remains +// useful only for gross-mismatch detection. var kvExtShadowTotal = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "dq_lake_latest_kv_ext_shadow_total", Help: "Shadow comparisons of the extended KV paths (allLatest|availableSignals) against the rollup (match|kv_newer|kv_miss|mismatch).", diff --git a/settings.sample.yaml b/settings.sample.yaml index eb5f8c5..495f1e4 100644 --- a/settings.sample.yaml +++ b/settings.sample.yaml @@ -42,13 +42,14 @@ MATERIALIZER_ROLLUP_INTERVAL: '' MATERIALIZER_BACKFILL_MODE: false # Read only to REJECT sharding (> 1 is refused — single writer only). MATERIALIZER_SHARD_COUNT: 1 -# Daily signals_latest refresh (dq#55): off | shadow | on. shadow maintains -# lake.signals_latest_daily by a once-daily watermarked fold and diffs it -# against the incrementally-folded rollup (the evidence gating the flip); on -# is the flip — the daily refresh maintains lake.signals_latest itself, the -# per-pass fold is off, and the first boot after shadow promotes the shadow -# table. Pair on with LAKE_ROLLUP_DAILY_SERVING=true on the query fleet. -MATERIALIZER_DAILY_ROLLUP_MODE: 'off' +# Daily signals_latest refresh (dq#55): off | on, default on (empty means on). +# on: the once-daily watermarked refresh is THE maintainer of +# lake.signals_latest (the per-pass fold was removed in step 5). off leaves +# the rollup unmaintained — tests/one-off ops only, warned at boot. The +# retired shadow mode is invalid; shadow-era configs must move to on (a +# leftover lake.signals_latest_daily is promoted automatically at first +# boot). Pair on with LAKE_ROLLUP_DAILY_SERVING=true on the query fleet. +MATERIALIZER_DAILY_ROLLUP_MODE: 'on' # How long after UTC midnight the daily refresh waits (cursor settle + # straggler margin). Empty = 3h30m, the traffic trough. MATERIALIZER_DAILY_ROLLUP_DELAY: '' @@ -56,9 +57,10 @@ MATERIALIZER_DAILY_ROLLUP_DELAY: '' # off | shadow | serve, dark-launched independently of LATEST_KV_READ_MODE # (which must not be off when this is set). Query fleet only. LATEST_KV_READ_MODE_EXTENDED: 'off' -# Set true ONLY at the dq#55 step-4 flip (rollup maintained by the daily -# refresh): summaries then serve the exact (rollup ∪ tail) union. Query fleet. -LAKE_ROLLUP_DAILY_SERVING: false +# True whenever the materializer runs MATERIALIZER_DAILY_ROLLUP_MODE=on (the +# default): summaries then serve the exact (rollup ∪ tail) union — a plain +# rollup read under-counts the post-watermark tail. Query fleet. +LAKE_ROLLUP_DAILY_SERVING: true # Decoded-row retention (Go duration, e.g. 8760h); empty disables pruning. LAKE_DECODED_RETENTION: "" # Rebuild signals_latest from full base on boot (disaster recovery only; O(history)). diff --git a/tests/dis_parity_test.go b/tests/dis_parity_test.go index fb6a46b..ed39185 100644 --- a/tests/dis_parity_test.go +++ b/tests/dis_parity_test.go @@ -173,6 +173,9 @@ func materializeRuptela(t *testing.T) *duck.Service { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Positive(t, drainRunner(t, ctx, runner)) + // signalsLatest is rollup-served; make it current through tomorrow's UTC + // midnight (covers the fixture's timestamp wherever it falls before now). + refreshRollup(t, ctx, mat, time.Now().UTC().Truncate(24*time.Hour).AddDate(0, 0, 1)) return svc } diff --git a/tests/ducklake_daily_flip_test.go b/tests/ducklake_daily_flip_test.go index a14a784..14fb3a8 100644 --- a/tests/ducklake_daily_flip_test.go +++ b/tests/ducklake_daily_flip_test.go @@ -1,35 +1,26 @@ -// ducklake_daily_flip_test.go proves dq#55 step 4: the shadow→on promote -// (including remediation of a duplicate-corrupted live table), the per-pass -// fold going quiet under mode on, and the daily refresh maintaining -// lake.signals_latest itself — still exactly equal to a full recompute over -// settled data. +// ducklake_daily_flip_test.go proves the dq#55 boot transitions of the daily +// refresh: the shadow→on promote (a leftover shadow-era +// lake.signals_latest_daily table found at boot under mode on is swapped into +// lake.signals_latest — including remediation of a duplicate-corrupted live +// table — then dropped forever), decode leaving lake.signals_latest untouched +// (the per-pass fold is gone, step 5), and the daily refresh maintaining the +// live table itself — still exactly equal to a full recompute over settled +// data. Shadow MODE no longer exists, so the shadow-era state is manufactured +// directly: the table, its content (a bounded recompute), and the watermark +// row — exactly what a node upgrading straight from mode=shadow boots with. package tests import ( "context" - "database/sql" "fmt" "testing" "time" "github.com/DIMO-Network/dq/internal/materializer" - "github.com/rs/zerolog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// oracleRecompute rebuilds lake.signals_latest from the full base with a -// mode-off materializer (RecomputeRollup is deliberately disabled under mode -// on) and returns the rows — the exactness oracle for flip tests, valid when -// every seeded row is stamped before the last refreshed boundary. -func oracleRecompute(t *testing.T, ctx context.Context, db *sql.DB) map[string]rollupRow { - t.Helper() - oracle, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) - require.NoError(t, err) - require.NoError(t, oracle.RecomputeRollup(ctx)) - return dumpRollupMap(t, ctx, db) -} - func TestDuckLake_DailyFlip_PromoteHealsAndServes(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -38,20 +29,24 @@ func TestDuckLake_DailyFlip_PromoteHealsAndServes(t *testing.T) { day0 := time.Now().UTC().AddDate(0, 0, -4).Truncate(24 * time.Hour) b1, b2 := day0.AddDate(0, 0, 1), day0.AddDate(0, 0, 2) - // Shadow era: fold on, shadow table seeded and folded once. - shadowRunner, shadowMat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) - }) - require.NoError(t, shadowMat.LoadDailyRollupState(ctx)) + // Base rows across two settled days, decoded by a default-mode writer. + preRunner, preMat := incrRunner(t, ctx, db) seedRawStatus(t, db, "fl-1", subj, day0.Add(time.Hour), speedAt(day0.Add(time.Hour), 30)) - drainNoFlush(t, ctx, shadowRunner) - require.NoError(t, shadowMat.RunDailyRollupRefresh(ctx, b1)) seedRawStatus(t, db, "fl-2", subj, b1.Add(time.Hour), speedAt(b1.Add(time.Hour), 44), odoAt(b1.Add(time.Hour), 500)) - drainNoFlush(t, ctx, shadowRunner) - require.NoError(t, shadowMat.RunDailyRollupRefresh(ctx, b2)) + drainNoFlush(t, ctx, preRunner) + + // Manufacture the shadow-era leftovers a straight-from-shadow upgrade boots + // with: a validated one-row-per-key shadow table exact through b2 (all data + // is < b2, so a full recompute IS the bounded one) and its watermark row. + require.NoError(t, preMat.RecomputeRollup(ctx)) + _, err := db.ExecContext(ctx, "CREATE TABLE lake.signals_latest_daily AS SELECT * FROM lake.signals_latest") + require.NoError(t, err) + _, err = db.ExecContext(ctx, "INSERT INTO lake.ingest_progress (partition, cursor) VALUES (?, ?)", + "lake.signals_latest#daily_watermark", b2.Format(time.RFC3339)) + require.NoError(t, err) - // Manufacture the production corruption: visible duplicate rows on the - // live table (the per-pass-fold-vs-compaction race residue). + // Manufacture the production corruption on the live table: visible + // duplicate rows (the per-pass-fold-vs-compaction race residue). for i := 0; i < 5; i++ { _, err := db.ExecContext(ctx, `INSERT INTO lake.signals_latest SELECT * FROM lake.signals_latest WHERE subject = ? AND name = 'speed' LIMIT 1`, subj) @@ -61,7 +56,7 @@ func TestDuckLake_DailyFlip_PromoteHealsAndServes(t *testing.T) { require.NoError(t, db.QueryRowContext(ctx, "SELECT count(*) FROM lake.signals_latest WHERE subject = ? AND name = 'speed'", subj).Scan(&liveRows)) require.Greater(t, liveRows, 1, "corruption manufactured") - // The flip: a new materializer in mode on. Loading state promotes the + // The flip boot: a new materializer in mode on. Loading state promotes the // shadow table — the corrupted live content is discarded wholesale. flipRunner, flipMat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithDailyRollup(materializer.DailyRollupOn, 0) @@ -82,7 +77,7 @@ func TestDuckLake_DailyFlip_PromoteHealsAndServes(t *testing.T) { seedRawStatus(t, db, "fl-3", subj, newTS, speedAt(newTS, 77)) drainNoFlush(t, ctx, flipRunner) afterDecode := dumpRollupMap(t, ctx, db) - assert.EqualValues(t, 2, afterDecode[subj+"|speed"].count, "per-pass fold is off: the rollup is untouched by decode") + assert.EqualValues(t, 2, afterDecode[subj+"|speed"].count, "per-pass fold is gone: the rollup is untouched by decode") assert.EqualValues(t, 44, afterDecode[subj+"|speed"].valueNumber.Float64) // The next refresh folds the tail into the LIVE table; the result must @@ -103,6 +98,37 @@ func TestDuckLake_DailyFlip_PromoteHealsAndServes(t *testing.T) { } } +// TestDuckLake_DailyFlip_AbortedShadowSeedIsDropped covers the degenerate +// leftover: a shadow table WITHOUT a watermark (an aborted shadow-era seed) is +// worthless as a promote source — boot under mode on must drop it, leave the +// live table alone, and let the first refresh seed lake.signals_latest. +func TestDuckLake_DailyFlip_AbortedShadowSeedIsDropped(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:123", vehicleNFT.Hex()) + day0 := time.Now().UTC().AddDate(0, 0, -2).Truncate(24 * time.Hour) + b1 := day0.AddDate(0, 0, 1) + + runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { + m.WithDailyRollup(materializer.DailyRollupOn, 0) + }) + seedRawStatus(t, db, "as-1", subj, day0.Add(time.Hour), speedAt(day0.Add(time.Hour), 41)) + drainNoFlush(t, ctx, runner) + // A shadow table with no watermark row: the aborted-seed leftover. + _, err := db.ExecContext(ctx, "CREATE TABLE lake.signals_latest_daily AS SELECT * FROM lake.signals_latest WHERE false") + require.NoError(t, err) + + require.NoError(t, mat.LoadDailyRollupState(ctx)) + var shadowExists int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*) FROM duckdb_tables() WHERE database_name = 'lake' AND table_name = 'signals_latest_daily'`).Scan(&shadowExists)) + assert.Zero(t, shadowExists, "unwatermarked shadow table dropped, not promoted") + + require.NoError(t, mat.RunDailyRollupRefresh(ctx, b1)) + assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subj+"|speed"].count, "first refresh seeds the live table") +} + // TestDuckLake_DailyFlip_FreshInstallSeedsLive covers mode on with no shadow // era at all: the first refresh seeds lake.signals_latest directly (bounded // recompute, no DROP of the serving table), and subsequent folds maintain it. diff --git a/tests/ducklake_daily_rollup_test.go b/tests/ducklake_daily_rollup_test.go index 3aa28f3..dd0c70a 100644 --- a/tests/ducklake_daily_rollup_test.go +++ b/tests/ducklake_daily_rollup_test.go @@ -1,11 +1,12 @@ -// ducklake_daily_rollup_test.go proves dq#55 step 1: the daily watermarked -// refresh maintains lake.signals_latest_daily EXACTLY equal to the -// incrementally-folded lake.signals_latest over settled data — across the -// seed, the daily fold (new names, new subjects, collisions, redeliveries, -// location fixes, multi-day catch-up), and the late-arrival path (rows -// stamped before the watermark arriving after it), with the shadow diff -// reporting zero. The live rollup is the oracle here; its own exactness -// against RecomputeRollup is proven by ducklake_incremental_rollup_test.go. +// ducklake_daily_rollup_test.go proves the daily watermarked refresh (dq#55) +// as the SOLE maintainer of lake.signals_latest (the per-pass fold was removed +// in step 5): the seed, the daily fold (new names, new subjects, collisions, +// redeliveries, location fixes, multi-day catch-up), refresh idempotency (an +// already-covered boundary must no-op, never double-fold), and the +// late-arrival path (rows stamped before the watermark arriving after it) — +// always column-identical to a full recompute over the deduped base +// (assertRollupMatchesOracle). The promote / fresh-install flip paths live in +// ducklake_daily_flip_test.go. package tests import ( @@ -20,58 +21,6 @@ import ( "github.com/stretchr/testify/require" ) -func dumpDailyMap(t *testing.T, ctx context.Context, db *sql.DB) map[string]rollupRow { - t.Helper() - rows, err := db.QueryContext(ctx, - `SELECT subject, name, subject_bucket, "timestamp", value_number, value_string, - loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen - FROM lake.signals_latest_daily ORDER BY subject, name`) - require.NoError(t, err) - defer rows.Close() //nolint:errcheck - out := map[string]rollupRow{} - for rows.Next() { - var r rollupRow - require.NoError(t, rows.Scan(&r.subject, &r.name, &r.bucket, &r.timestamp, &r.valueNumber, &r.valueString, - &r.locLat, &r.locLon, &r.locHdop, &r.locHeading, &r.locTS, &r.count, &r.firstSeen, &r.lastSeen)) - out[r.subject+"|"+r.name] = r - } - require.NoError(t, rows.Err()) - return out -} - -// assertDailyMatchesLive compares the daily table against the live rollup, -// column by column. Valid whenever every seeded row's timestamp is before the -// last refreshed boundary (all data settled), which every scenario below -// arranges — then the two tables must be identical. -func assertDailyMatchesLive(t *testing.T, ctx context.Context, db *sql.DB) { - t.Helper() - live := dumpRollupMap(t, ctx, db) - daily := dumpDailyMap(t, ctx, db) - keys := map[string]bool{} - for k := range live { - keys[k] = true - } - for k := range daily { - keys[k] = true - } - for k := range keys { - l, okL := live[k] - d, okD := daily[k] - require.Truef(t, okL, "%s in the daily table but MISSING from the live rollup", k) - require.Truef(t, okD, "%s in the live rollup but MISSING from the daily table", k) - assert.Equalf(t, l.count, d.count, "%s count", k) - assert.Truef(t, l.timestamp.Equal(d.timestamp), "%s timestamp: live=%s daily=%s", k, l.timestamp, d.timestamp) - assert.Equalf(t, l.valueNumber, d.valueNumber, "%s value_number", k) - assert.Equalf(t, l.valueString, d.valueString, "%s value_string", k) - assert.Truef(t, l.firstSeen.Equal(d.firstSeen), "%s first_seen: live=%s daily=%s", k, l.firstSeen, d.firstSeen) - assert.Truef(t, l.lastSeen.Equal(d.lastSeen), "%s last_seen: live=%s daily=%s", k, l.lastSeen, d.lastSeen) - assert.InDeltaf(t, l.locLat, d.locLat, 1e-9, "%s loc_lat", k) - assert.InDeltaf(t, l.locLon, d.locLon, 1e-9, "%s loc_lon", k) - assert.InDeltaf(t, l.locHdop, d.locHdop, 1e-9, "%s loc_hdop", k) - assert.Truef(t, l.locTS.Equal(d.locTS), "%s loc_ts: live=%s daily=%s", k, l.locTS, d.locTS) - } -} - func locFixAt(ts time.Time, lat, lon float64) map[string]any { return map[string]any{"name": "currentLocationCoordinates", "timestamp": ts.Format(time.RFC3339Nano), "value": map[string]any{"latitude": lat, "longitude": lon, "hdop": 1.5}} @@ -85,13 +34,6 @@ func storedWatermark(t *testing.T, ctx context.Context, db *sql.DB) string { return w } -func requireZeroDiff(t *testing.T, counts map[string]int) { - t.Helper() - for class, n := range counts { - assert.Zerof(t, n, "shadow diff class %s", class) - } -} - func TestDuckLake_DailyRollup_SeedFoldAndLateArrivals(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -104,7 +46,7 @@ func TestDuckLake_DailyRollup_SeedFoldAndLateArrivals(t *testing.T) { b1, b2, b4 := day0.AddDate(0, 0, 1), day0.AddDate(0, 0, 2), day0.AddDate(0, 0, 4) runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) + m.WithDailyRollup(materializer.DailyRollupOn, 0) }) require.NoError(t, mat.LoadDailyRollupState(ctx)) @@ -117,20 +59,21 @@ func TestDuckLake_DailyRollup_SeedFoldAndLateArrivals(t *testing.T) { seedRawStatus(t, db, "d0-4", s1, collTS, speedAt(collTS, 31)) // same (s,n,ts), different ceid seedRawStatus(t, db, "d0-5", s2, day0.Add(1*time.Hour), speedAt(day0.Add(1*time.Hour), 55)) drainNoFlush(t, ctx, runner) + assert.Empty(t, dumpRollupMap(t, ctx, db), "nothing maintains the rollup at decode time (the fold is gone)") - // Seed: the daily table becomes the bounded recompute over ts < b1. + // Seed: lake.signals_latest becomes the bounded recompute over ts < b1. require.NoError(t, mat.RunDailyRollupRefresh(ctx, b1)) - assertDailyMatchesLive(t, ctx, db) + assertRollupMatchesOracle(t, ctx, db) assert.Equal(t, b1.Format(time.RFC3339), storedWatermark(t, ctx, db), "watermark persisted at the boundary") - counts, err := mat.DailyRollupDiff(ctx) - require.NoError(t, err) - requireZeroDiff(t, counts) + assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[s1+"|speed"].count, + "same-(subject,name,timestamp) collision is one distinct reading: 3 counted, not 4") // Guards: a non-midnight boundary is refused; an already-covered boundary - // is a no-op that must not move the watermark backwards. + // is a no-op that must not move the watermark backwards or double-fold. require.Error(t, mat.RunDailyRollupRefresh(ctx, b1.Add(90*time.Minute))) require.NoError(t, mat.RunDailyRollupRefresh(ctx, b1)) assert.Equal(t, b1.Format(time.RFC3339), storedWatermark(t, ctx, db)) + assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[s1+"|speed"].count, "re-running a covered boundary changes nothing") // Day 1: a new name for s1, a brand-new subject s3, a newer fix, and a // redelivery of a day-0 event (arrives late → dedup'd in base, late-marked, @@ -143,10 +86,7 @@ func TestDuckLake_DailyRollup_SeedFoldAndLateArrivals(t *testing.T) { drainNoFlush(t, ctx, runner) require.NoError(t, mat.RunDailyRollupRefresh(ctx, b2)) - assertDailyMatchesLive(t, ctx, db) - counts, err = mat.DailyRollupDiff(ctx) - require.NoError(t, err) - requireZeroDiff(t, counts) + assertRollupMatchesOracle(t, ctx, db) // Days 2-3, folded in ONE refresh (a missed boundary catches up), plus a // genuinely late NEW reading: s2 uploads a buffered day-0 row AFTER the @@ -158,24 +98,20 @@ func TestDuckLake_DailyRollup_SeedFoldAndLateArrivals(t *testing.T) { drainNoFlush(t, ctx, runner) require.NoError(t, mat.RunDailyRollupRefresh(ctx, b4)) - assertDailyMatchesLive(t, ctx, db) - counts, err = mat.DailyRollupDiff(ctx) - require.NoError(t, err) - requireZeroDiff(t, counts) - assert.EqualValues(t, 3, dumpDailyMap(t, ctx, db)[s2+"|speed"].count, + assertRollupMatchesOracle(t, ctx, db) + assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[s2+"|speed"].count, "the late buffered reading is counted after the late-set recompute") var lateLeft int require.NoError(t, db.QueryRowContext(ctx, "SELECT count(*) FROM lake.rollup_late_subjects").Scan(&lateLeft)) assert.Zero(t, lateLeft, "late marks cleared by the recompute that covered them") - // The diff actually detects divergence: corrupt one daily row and expect a - // mismatch, proving zero-diff above is a real assertion, not vacuity. - _, err = db.ExecContext(ctx, - "UPDATE lake.signals_latest_daily SET count = count + 1 WHERE subject = ? AND name = 'speed'", s1) - require.NoError(t, err) - counts, err = mat.DailyRollupDiff(ctx) - require.NoError(t, err) - assert.Equal(t, 1, counts["mismatch"], "a corrupted daily row must surface as a mismatch") + // One row per (subject, name), always — the cardinality contract the + // per-pass fold used to violate under compaction races. + var keys, rows int + require.NoError(t, db.QueryRowContext(ctx, + `SELECT count(*), coalesce(sum(n), 0) FROM (SELECT count(*) AS n FROM lake.signals_latest GROUP BY subject, name)`). + Scan(&keys, &rows)) + assert.Equal(t, keys, rows, "exactly one physical rollup row per key") } // TestDuckLake_DailyRollup_WatermarkSurvivesRestart proves the watermark (and @@ -191,7 +127,7 @@ func TestDuckLake_DailyRollup_WatermarkSurvivesRestart(t *testing.T) { b1, b2 := day0.AddDate(0, 0, 1), day0.AddDate(0, 0, 2) runner1, mat1 := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) + m.WithDailyRollup(materializer.DailyRollupOn, 0) }) require.NoError(t, mat1.LoadDailyRollupState(ctx)) seedRawStatus(t, db, "w1", subj, day0.Add(5*time.Minute), speedAt(day0.Add(5*time.Minute), 12)) @@ -200,7 +136,7 @@ func TestDuckLake_DailyRollup_WatermarkSurvivesRestart(t *testing.T) { // "Restart": a new materializer over the same catalog. runner2, mat2 := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) + m.WithDailyRollup(materializer.DailyRollupOn, 0) }) require.NoError(t, mat2.LoadDailyRollupState(ctx)) require.NoError(t, mat2.RunDailyRollupRefresh(ctx, b1), "already-covered boundary must no-op on the restarted writer") @@ -209,6 +145,6 @@ func TestDuckLake_DailyRollup_WatermarkSurvivesRestart(t *testing.T) { seedRawStatus(t, db, "w2", subj, b1.Add(10*time.Minute), speedAt(b1.Add(10*time.Minute), 24)) drainNoFlush(t, ctx, runner2) require.NoError(t, mat2.RunDailyRollupRefresh(ctx, b2)) - assertDailyMatchesLive(t, ctx, db) - assert.EqualValues(t, 2, dumpDailyMap(t, ctx, db)[subj+"|speed"].count) + assertRollupMatchesOracle(t, ctx, db) + assert.EqualValues(t, 2, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) } diff --git a/tests/ducklake_incremental_rollup_test.go b/tests/ducklake_incremental_rollup_test.go deleted file mode 100644 index 3966f52..0000000 --- a/tests/ducklake_incremental_rollup_test.go +++ /dev/null @@ -1,309 +0,0 @@ -// ducklake_incremental_rollup_test.go proves finding #5b: lake.signals_latest is -// maintained INCREMENTALLY at decode-commit time (O(batch), not an O(history) recompute -// per flush), and the incremental result is EXACTLY the full RecomputeRollup — the -// differential invariant, checked across redelivery, same-timestamp collision, -// out-of-order arrival, multi-window (#1c) spans, location updates, and dormant re-report. -package tests - -import ( - "context" - "database/sql" - "fmt" - "math/rand" - "sort" - "testing" - "time" - - "github.com/DIMO-Network/dq/internal/materializer" - "github.com/rs/zerolog" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -type rollupRow struct { - subject, name string - bucket int - timestamp time.Time - valueNumber sql.NullFloat64 - valueString sql.NullString - locLat, locLon float64 - locHdop, locHeading float64 - locTS time.Time - count int64 - firstSeen, lastSeen time.Time -} - -func dumpRollupMap(t *testing.T, ctx context.Context, db *sql.DB) map[string]rollupRow { - t.Helper() - rows, err := db.QueryContext(ctx, - `SELECT subject, name, subject_bucket, "timestamp", value_number, value_string, - loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen - FROM lake.signals_latest ORDER BY subject, name`) - require.NoError(t, err) - defer rows.Close() //nolint:errcheck - out := map[string]rollupRow{} - for rows.Next() { - var r rollupRow - require.NoError(t, rows.Scan(&r.subject, &r.name, &r.bucket, &r.timestamp, &r.valueNumber, &r.valueString, - &r.locLat, &r.locLon, &r.locHdop, &r.locHeading, &r.locTS, &r.count, &r.firstSeen, &r.lastSeen)) - out[r.subject+"|"+r.name] = r - } - require.NoError(t, rows.Err()) - return out -} - -// assertMatchesRecompute snapshots the incrementally-maintained rollup, rebuilds it from -// scratch with RecomputeRollup, and asserts every column of every (subject,name) row is -// identical. This is the exactness contract for #5b. -func assertMatchesRecompute(t *testing.T, ctx context.Context, db *sql.DB, mat *materializer.DuckLakeMaterializer) { - t.Helper() - incremental := dumpRollupMap(t, ctx, db) - require.NoError(t, mat.RecomputeRollup(ctx)) - recomputed := dumpRollupMap(t, ctx, db) - - keys := map[string]bool{} - for k := range incremental { - keys[k] = true - } - for k := range recomputed { - keys[k] = true - } - sorted := make([]string, 0, len(keys)) - for k := range keys { - sorted = append(sorted, k) - } - sort.Strings(sorted) - for _, k := range sorted { - inc, okI := incremental[k] - rec, okR := recomputed[k] - require.Truef(t, okI, "%s present after RecomputeRollup but MISSING from the incremental rollup", k) - require.Truef(t, okR, "%s present in the incremental rollup but MISSING after RecomputeRollup", k) - assert.Equalf(t, rec.count, inc.count, "%s count", k) - assert.Truef(t, rec.timestamp.Equal(inc.timestamp), "%s timestamp: recompute=%s incremental=%s", k, rec.timestamp, inc.timestamp) - assert.Equalf(t, rec.valueNumber, inc.valueNumber, "%s value_number", k) - assert.Equalf(t, rec.valueString, inc.valueString, "%s value_string", k) - assert.Truef(t, rec.firstSeen.Equal(inc.firstSeen), "%s first_seen: recompute=%s incremental=%s", k, rec.firstSeen, inc.firstSeen) - assert.Truef(t, rec.lastSeen.Equal(inc.lastSeen), "%s last_seen: recompute=%s incremental=%s", k, rec.lastSeen, inc.lastSeen) - assert.InDeltaf(t, rec.locLat, inc.locLat, 1e-9, "%s loc_lat", k) - assert.InDeltaf(t, rec.locLon, inc.locLon, 1e-9, "%s loc_lon", k) - assert.Truef(t, rec.locTS.Equal(inc.locTS), "%s loc_ts: recompute=%s incremental=%s", k, rec.locTS, inc.locTS) - } -} - -func incrRunner(t *testing.T, ctx context.Context, db *sql.DB, opts ...func(*materializer.DuckLakeMaterializer)) (*materializer.Runner, *materializer.DuckLakeMaterializer) { - t.Helper() - mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) - require.NoError(t, err) - for _, o := range opts { - o(mat) - } - runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) - return runner, mat -} - -// drainNoFlush runs the decode loop to completion WITHOUT calling FlushRollup, so -// lake.signals_latest is populated ONLY by the commit-time incremental fold (#5b) — the -// whole point of the differential test. -func drainNoFlush(t *testing.T, ctx context.Context, r *materializer.Runner) { - t.Helper() - for { - n, err := r.RunOnce(ctx) - require.NoError(t, err) - if n == 0 { - return - } - } -} - -// TestDuckLake_IncrementalRollup_MatchesRecompute drives the decode loop WITHOUT ever -// calling FlushRollup, so lake.signals_latest is populated only by the commit-time -// incremental fold, then asserts it equals a full RecomputeRollup after each scenario. -func TestDuckLake_IncrementalRollup_MatchesRecompute(t *testing.T) { - ctx := context.Background() - subject := fmt.Sprintf("did:erc721:137:%s:71", vehicleNFT.Hex()) - base := time.Now().UTC().AddDate(0, 0, -3).Truncate(time.Hour) - - t.Run("basic multi-batch increasing", func(t *testing.T) { - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - runner, mat := incrRunner(t, ctx, db) - seedRawStatus(t, db, "b1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) - seedRawStatus(t, db, "b2", subject, base.Add(2*time.Minute), speedAt(base.Add(2*time.Minute), 20)) - drainNoFlush(t, ctx, runner) - seedRawStatus(t, db, "b3", subject, base.Add(3*time.Minute), speedAt(base.Add(3*time.Minute), 30)) - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[subject+"|speed"].count) - }) - - t.Run("redelivery same cloud_event_id no double count", func(t *testing.T) { - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - runner, mat := incrRunner(t, ctx, db) - seedRawStatus(t, db, "r1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) - drainNoFlush(t, ctx, runner) - seedRawStatus(t, db, "r1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) // redelivery - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subject+"|speed"].count) - }) - - t.Run("same-timestamp collision distinct count", func(t *testing.T) { - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - runner, mat := incrRunner(t, ctx, db) - ts := base.Add(5 * time.Minute) - seedRawStatus(t, db, "c1", subject, ts, speedAt(ts, 40)) - seedRawStatus(t, db, "c2", subject, ts, speedAt(ts, 41)) // same (s,n,ts), different ceid - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "a same-(subject,name,timestamp) collision is one distinct row") - }) - - t.Run("out-of-order older batch", func(t *testing.T) { - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - runner, mat := incrRunner(t, ctx, db) - seedRawStatus(t, db, "o2", subject, base.Add(10*time.Minute), speedAt(base.Add(10*time.Minute), 80)) - drainNoFlush(t, ctx, runner) - seedRawStatus(t, db, "o1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 5)) // older, arrives later - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - got := dumpRollupMap(t, ctx, db)[subject+"|speed"] - assert.EqualValues(t, 2, got.count) - assert.EqualValues(t, 80, got.valueNumber.Float64, "latest stays the newer reading despite the later-arriving older one") - assert.True(t, got.firstSeen.Equal(base.Add(1*time.Minute)), "first_seen moves back to the older reading") - }) - - t.Run("multi-window fat span exact", func(t *testing.T) { - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(3) }) - seedRawStatusOneSnapshot(t, db, subject, base.Add(20*time.Minute), 10) // one snapshot, 10 rows, 3/window - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - assert.EqualValues(t, 10, dumpRollupMap(t, ctx, db)[subject+"|speed"].count) - }) - - t.Run("dormant re-report", func(t *testing.T) { - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - runner, mat := incrRunner(t, ctx, db) - seedRawStatus(t, db, "d1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) - drainNoFlush(t, ctx, runner) - // long gap, then report again - seedRawStatus(t, db, "d2", subject, base.Add(48*time.Hour), speedAt(base.Add(48*time.Hour), 55)) - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - got := dumpRollupMap(t, ctx, db)[subject+"|speed"] - assert.EqualValues(t, 2, got.count) - assert.EqualValues(t, 55, got.valueNumber.Float64) - }) -} - -// TestDuckLake_IncrementalRollup_CrashReplayExact proves the incremental fold is -// idempotent across a crash: a window commits its base rows AND its rollup delta, the pass -// then crashes before finishing the span, and on restart the replayed window must add 0 to -// count (rows already at rest → NOT-EXISTS delta 0, recency re-fold stable), so the final -// rollup still equals a full RecomputeRollup. -func TestDuckLake_IncrementalRollup_CrashReplayExact(t *testing.T) { - ctx := context.Background() - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - subject := fmt.Sprintf("did:erc721:137:%s:72", vehicleNFT.Hex()) - base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) - seedRawStatusOneSnapshot(t, db, subject, base, 9) // one snapshot, 9 rows - - // mat1 crashes right after the first intermediate window commits (base + rollup delta). - mat1, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) - require.NoError(t, err) - mat1.WithMaxRowsPerWindow(3) - mat1.WithWindowCommitHook(func(idx int) error { - if idx == 0 { - return fmt.Errorf("injected crash after first window") - } - return nil - }) - runner1 := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat1) - _, err = runner1.RunOnce(ctx) - require.Error(t, err) - assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "first window's rollup delta is durable") - - // Restart and drain: the replayed window must not double-count. - runner2, mat2 := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(3) }) - _ = runner2 - drainNoFlush(t, ctx, runner2) - assert.EqualValues(t, 9, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "replayed window added 0 — exactly nine") - assertMatchesRecompute(t, ctx, db, mat2) -} - -// TestDuckLake_IncrementalRollup_Randomized fuzzes many batches (redeliveries, collisions, -// out-of-order, varied window sizes, multiple subjects/names, location fixes) through the -// incremental path and asserts the result equals RecomputeRollup every time. Deterministic -// seed for reproducibility. -func TestDuckLake_IncrementalRollup_Randomized(t *testing.T) { - ctx := context.Background() - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - rng := rand.New(rand.NewSource(20260707)) - subjects := []string{ - fmt.Sprintf("did:erc721:137:%s:81", vehicleNFT.Hex()), - fmt.Sprintf("did:erc721:137:%s:82", vehicleNFT.Hex()), - fmt.Sprintf("did:erc721:137:%s:83", vehicleNFT.Hex()), - } - names := []func(ts time.Time, v float64) map[string]any{speedAt, odoAt} - base := time.Now().UTC().AddDate(0, 0, -5).Truncate(time.Hour) - - runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(2 + rng.Intn(5)) }) - seen := map[string]bool{} // ids emitted, to force redeliveries - seq := 0 - for round := 0; round < 25; round++ { - n := 1 + rng.Intn(4) - for i := 0; i < n; i++ { - subj := subjects[rng.Intn(len(subjects))] - nameFn := names[rng.Intn(len(names))] - // timestamps wander forward and sometimes backward (out-of-order) - off := time.Duration(seq*7+rng.Intn(400)-100) * time.Second - ts := base.Add(off) - seq++ - id := fmt.Sprintf("rnd-%d", seq) - if len(seen) > 0 && rng.Intn(5) == 0 { // redelivery of a prior id - for k := range seen { - id = k - break - } - } else if rng.Intn(6) == 0 && seq > 1 { // collision: new id, reuse a recent timestamp - ts = base.Add(time.Duration((seq-1)*7) * time.Second) - } - seen[id] = true - seedRawStatus(t, db, id, subj, ts, nameFn(ts, float64(rng.Intn(120)))) - } - drainNoFlush(t, ctx, runner) - if round%5 == 4 { - assertMatchesRecompute(t, ctx, db, mat) - } - } - assertMatchesRecompute(t, ctx, db, mat) -} - -func odoAt(ts time.Time, v float64) map[string]any { - return map[string]any{"name": "powertrainTransmissionTravelledDistance", "timestamp": ts.Format(time.RFC3339Nano), "value": v} -} - -// TestDuckLake_IncrementalRollup_AncientRedelivery pins the >30d-redelivery count fix: a -// reading older than the 30d dedup probe floor, redelivered (same cloud_event_id) through -// the live path, must NOT inflate signals_latest.count — it stays equal to RecomputeRollup. -func TestDuckLake_IncrementalRollup_AncientRedelivery(t *testing.T) { - ctx := context.Background() - svc := newLakeService(t, t.TempDir()) - db := svc.DB() - subj := fmt.Sprintf("did:erc721:137:%s:73", vehicleNFT.Hex()) - ancient := time.Now().UTC().Add(-40 * 24 * time.Hour).Truncate(time.Hour) // > dedupProbeFloor (30d) - runner, mat := incrRunner(t, ctx, db) - seedRawStatus(t, db, "anc1", subj, ancient, speedAt(ancient, 22)) - drainNoFlush(t, ctx, runner) - seedRawStatus(t, db, "anc1", subj, ancient, speedAt(ancient, 22)) // redelivery of the SAME event, >30d old - drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) - assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subj+"|speed"].count, "an ancient redelivery must not inflate count") -} diff --git a/tests/ducklake_latest_rollup_test.go b/tests/ducklake_latest_rollup_test.go index 423fd9b..c874aa2 100644 --- a/tests/ducklake_latest_rollup_test.go +++ b/tests/ducklake_latest_rollup_test.go @@ -35,6 +35,7 @@ func TestDuckLake_GetLatestSignals_ServedFromRollup(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 2, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) // Drop the base history; the rollup is current state and stays (this is // exactly what PruneDecoded does at the retention boundary). @@ -63,7 +64,9 @@ func TestDuckLake_GetLatestSignals_ServedFromRollup(t *testing.T) { // and a later batch touching one subject refreshes that subject's rollup row // without disturbing (or depending on re-scanning) the others. Bucket-scoped // dirtiness saturated at trivial fleet activity and made every flush a -// full-table recompute on the decode goroutine. +// full-table recompute on the decode goroutine. Since dq#55 step 5 signal +// subjects are dirtied only in backfill mode (the bulk catch-up is the +// machinery's remaining consumer), so that is how this test drives it. func TestDuckLake_FlushRollup_SubjectScoped(t *testing.T) { ctx := context.Background() dir := t.TempDir() @@ -82,6 +85,7 @@ func TestDuckLake_FlushRollup_SubjectScoped(t *testing.T) { mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) require.NoError(t, err) + mat.WithBackfillMode(true) // the dirty-set flush's remaining consumer (see the test doc) runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 3, drainRunner(t, ctx, runner)) @@ -141,6 +145,7 @@ func TestDuckLake_LocationLatest_ServedFromRollup(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 1, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) // Retention prunes the base; the rollup is current state and stays. _, err = db.ExecContext(ctx, "DELETE FROM lake.signals") @@ -168,7 +173,9 @@ func TestDuckLake_LocationLatest_ServedFromRollup(t *testing.T) { // catch-up bound: when more distinct subjects dirty the rollup than the cap // allows (initial backfill defers the flush until fully drained), the dirty // set must not grow unbounded — FlushRollup escalates to the bucket-chunked -// full rebuild and the rollup still comes out complete and correct. +// full rebuild and the rollup still comes out complete and correct. Backfill +// mode is what dirties signal subjects since dq#55 step 5, so it drives the +// overflow here. func TestDuckLake_DirtySetOverflow_EscalatesToFullRebuild(t *testing.T) { ctx := context.Background() dir := t.TempDir() @@ -185,6 +192,7 @@ func TestDuckLake_DirtySetOverflow_EscalatesToFullRebuild(t *testing.T) { mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) require.NoError(t, err) + mat.WithBackfillMode(true) // signal subjects are dirtied only in backfill mode mat.WithMaxDirtySubjects(2) // force the overflow path with a tiny cap runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) diff --git a/tests/ducklake_migration_test.go b/tests/ducklake_migration_test.go index da43b20..41cfc0c 100644 --- a/tests/ducklake_migration_test.go +++ b/tests/ducklake_migration_test.go @@ -53,6 +53,10 @@ func TestDuckLake_LocTSMigration_ExistingCatalog(t *testing.T) { r0 := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat0) require.Equal(t, 1, drainRunner(t, ctx, r0)) + // The pre-H9 materializer maintained the rollup at decode time; today + // nothing does, so materialize the pre-migration rollup row explicitly + // before stripping loc_ts back off. + require.NoError(t, mat0.RecomputeRollup(ctx)) _, err = db.ExecContext(ctx, `ALTER TABLE lake.signals_latest DROP COLUMN loc_ts`) require.NoError(t, err) } diff --git a/tests/ducklake_only_test.go b/tests/ducklake_only_test.go index 1166900..51c9a79 100644 --- a/tests/ducklake_only_test.go +++ b/tests/ducklake_only_test.go @@ -216,6 +216,7 @@ func TestDuckLakeOnly_SegmentsSucceed(t *testing.T) { processed := drainRunner(t, ctx, runner) require.Equal(t, 2, processed, "two raw events decoded") + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) // LakeQueries serves GetAvailableSignals. lakeQ := duck.NewLakeQueries(lkSvc) diff --git a/tests/ducklake_pg_test.go b/tests/ducklake_pg_test.go index afa586b..c9ef21b 100644 --- a/tests/ducklake_pg_test.go +++ b/tests/ducklake_pg_test.go @@ -197,14 +197,14 @@ func TestDuckLakePostgres_ConcurrentPaginatedFatSnapshot(t *testing.T) { GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) assert.Zero(t, dupes, "no duplicate decoded rows across paginated windows") - // The commit-time incremental rollup (#5b) must be exact even though it ran across - // intermediate windows under two racing writers: it equals a full RecomputeRollup. - incremental := dumpRollupMap(t, ctx, db) - assert.EqualValues(t, events, incremental[subject+"|speed"].count, "incremental rollup count is exact under concurrent paginated writers") + // The rollup recompute over the racing writers' output must be exact: every + // row present once, so count equals the seeded events and recency is the + // newest reading. (The commit-time incremental fold this once compared + // against was removed in dq#55 step 5.) mat, err := materializer.NewDuckLakeMaterializer(ctx, newPGLakeService(t, dsn, dataPath).DB(), zerolog.Nop()) require.NoError(t, err) require.NoError(t, mat.RecomputeRollup(ctx)) recomputed := dumpRollupMap(t, ctx, db) - assert.Equal(t, recomputed[subject+"|speed"].count, incremental[subject+"|speed"].count, "incremental rollup == recompute (PG)") - assert.True(t, recomputed[subject+"|speed"].timestamp.Equal(incremental[subject+"|speed"].timestamp), "incremental recency == recompute (PG)") + assert.EqualValues(t, events, recomputed[subject+"|speed"].count, "rollup recompute count is exact under concurrent paginated writers") + assert.True(t, recomputed[subject+"|speed"].timestamp.Equal(base.Add(time.Duration(events-1)*time.Second).UTC()), "recency is the newest seeded reading (PG)") } diff --git a/tests/ducklake_query_test.go b/tests/ducklake_query_test.go index 1a8f4cb..d160298 100644 --- a/tests/ducklake_query_test.go +++ b/tests/ducklake_query_test.go @@ -35,6 +35,7 @@ func TestDuckLake_QueryBackend(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Positive(t, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) lake := duck.NewLakeQueries(svc) diff --git a/tests/ducklake_retention_test.go b/tests/ducklake_retention_test.go index 4cf5f00..0d079cc 100644 --- a/tests/ducklake_retention_test.go +++ b/tests/ducklake_retention_test.go @@ -65,6 +65,7 @@ func TestDuckLake_PruneDecoded_RemovesOrphanRollup(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 1, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, time.Now().UTC().Truncate(24*time.Hour)) rollupCount := func() int { var n int diff --git a/tests/ducklake_rollup_rebuild_test.go b/tests/ducklake_rollup_rebuild_test.go index e611364..299e346 100644 --- a/tests/ducklake_rollup_rebuild_test.go +++ b/tests/ducklake_rollup_rebuild_test.go @@ -1,8 +1,8 @@ // ducklake_rollup_rebuild_test.go covers the disaster-recovery rebuild of -// lake.signals_latest (RecomputeRollup): the per-batch refreshRollup only touches -// subjects present in a batch, so a dropped/truncated rollup needs a full rebuild -// from the base to repopulate dormant vehicles. RecomputeRollup must produce a -// rollup byte-identical to what the per-batch recompute built. +// lake.signals_latest (RecomputeRollup): the daily refresh only folds forward +// from its watermark, so a dropped/truncated rollup needs a full rebuild from +// the base to repopulate dormant vehicles. RecomputeRollup must produce a +// rollup byte-identical to what the daily refresh built. package tests import ( @@ -56,8 +56,9 @@ func TestRecomputeRollup_RebuildsDroppedRollupFromBase(t *testing.T) { seedRawStatus(t, db, "rr2", subjA, base.Add(2*time.Hour), speedAt(base.Add(2*time.Hour), 80)) seedRawStatus(t, db, "rr3", subjB, base.Add(time.Hour), speedAt(base.Add(time.Hour), 12)) require.Positive(t, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, time.Now().UTC().Truncate(24*time.Hour)) - // The rollup the per-batch recompute built. + // The rollup the daily refresh built. perBatch := dumpRollup(t, ctx, db) require.Len(t, perBatch, 2, "one (subject,name) row per vehicle") @@ -68,9 +69,13 @@ func TestRecomputeRollup_RebuildsDroppedRollupFromBase(t *testing.T) { require.NoError(t, err) require.Empty(t, dumpRollup(t, ctx, db)) - // Full rebuild from the base. - require.NoError(t, mat.RecomputeRollup(ctx)) + // Full rebuild from the base — via a default-mode materializer, since + // RecomputeRollup is refused under mode on (the reseed is the mode-on + // equivalent; DR keeps the unbounded rebuild). + oracle, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + require.NoError(t, oracle.RecomputeRollup(ctx)) require.Equal(t, perBatch, dumpRollup(t, ctx, db), - "RecomputeRollup must rebuild signals_latest byte-identical to the per-batch recompute") + "RecomputeRollup must rebuild signals_latest byte-identical to the daily refresh's output") } diff --git a/tests/ducklake_rollup_test.go b/tests/ducklake_rollup_test.go index fdfd750..d89fc3b 100644 --- a/tests/ducklake_rollup_test.go +++ b/tests/ducklake_rollup_test.go @@ -36,6 +36,7 @@ func TestDuckLake_LatestSummaryRollup(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 3, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) q := duck.NewLakeQueries(svc) @@ -68,9 +69,12 @@ func TestDuckLake_LatestSummaryRollup(t *testing.T) { require.NoError(t, err) assert.Contains(t, avail, "speed") - // Incremental: a fourth, newer reading updates the rollup latest to 90. + // Incremental: a fourth reading lands (stamped before the watermark, so it + // travels the late-subject path) and the NEXT refresh updates the rollup + // latest to 90. seedRawStatus(t, db, "rl-4", subject, day.Add(4*time.Hour), speedAt(day.Add(4*time.Hour), 90)) require.Equal(t, 1, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 2)) latest2, err := q.GetAllLatestSignals(ctx, subject, nil) require.NoError(t, err) for _, s := range latest2 { diff --git a/tests/ducklake_verify_pg_test.go b/tests/ducklake_verify_pg_test.go index 27e3d08..923b44e 100644 --- a/tests/ducklake_verify_pg_test.go +++ b/tests/ducklake_verify_pg_test.go @@ -1,6 +1,7 @@ -// ducklake_verify_pg_test.go — verification loops 9-10: the #1c pagination + #5b -// incremental rollup under real-Postgres concurrency with MORE than two writers and a -// mid-span crash of one writer while the others drain. Gated on PG_CATALOG_DSN. +// ducklake_verify_pg_test.go — verification loops 9-10: the #1c pagination +// exactly-once path under real-Postgres concurrency with MORE than two writers and a +// mid-span crash of one writer while the others drain, verified through a full +// rollup recompute over the deduped base. Gated on PG_CATALOG_DSN. package tests import ( @@ -18,7 +19,7 @@ import ( ) // V9 — three independent materializers drain the same paginated fat snapshot; exactly-once -// base rows AND an exact incremental rollup. +// base rows AND an exact rollup recompute over them. func TestVerify09_PG_ThreeConcurrentPaginated(t *testing.T) { dsn := pgCatalogDSN(t) ctx := context.Background() @@ -84,12 +85,12 @@ func TestVerify09_PG_ThreeConcurrentPaginated(t *testing.T) { require.NoError(t, db.QueryRowContext(ctx, `SELECT count(*) FROM ( SELECT cloud_event_id, name, timestamp FROM lake.signals GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) assert.Zero(t, dupes) - assert.EqualValues(t, events, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "incremental rollup exact under three writers") + assert.EqualValues(t, events, oracleRecompute(t, ctx, db)[subject+"|speed"].count, "rollup recompute exact under three writers") } // V10 — one writer crashes mid-span (its intermediate window errors) while two others drain; -// the idempotent windows + cursor-coupled final commit still yield exactly-once + an exact -// rollup. +// the idempotent windows + cursor-coupled final commit still yield exactly-once base rows +// (and so an exact rollup recompute). func TestVerify10_PG_CrashOneWriterMidSpan(t *testing.T) { dsn := pgCatalogDSN(t) ctx := context.Background() @@ -190,5 +191,5 @@ func TestVerify10_PG_CrashOneWriterMidSpan(t *testing.T) { require.NoError(t, db.QueryRowContext(ctx, `SELECT count(*) FROM ( SELECT cloud_event_id, name, timestamp FROM lake.signals GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) assert.Zero(t, dupes) - assert.EqualValues(t, events, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "incremental rollup exact after a mid-span crash under concurrency") + assert.EqualValues(t, events, oracleRecompute(t, ctx, db)[subject+"|speed"].count, "rollup recompute exact after a mid-span crash under concurrency") } diff --git a/tests/ducklake_verify_test.go b/tests/ducklake_verify_test.go index 61f3a02..7864c7a 100644 --- a/tests/ducklake_verify_test.go +++ b/tests/ducklake_verify_test.go @@ -1,7 +1,8 @@ -// ducklake_verify_test.go — adversarial verification campaign for the #1c pagination + -// #5b incremental-rollup exactly-once path. Each subtest is a distinct attack vector; the -// contract is always the same: the incrementally-maintained lake.signals_latest equals a -// full RecomputeRollup, and base rows are exactly-once. These are throwaway-hardening +// ducklake_verify_test.go — adversarial verification campaign for the #1c pagination +// exactly-once path. Each subtest is a distinct attack vector; the contract is always +// the same: base rows are exactly-once, observed through an explicit RecomputeRollup +// over the deduped base (the per-pass incremental fold was removed in dq#55 step 5, +// so the rollup is only ever recompute-derived here). These are throwaway-hardening // tests that also stay as regression guards. package tests @@ -22,8 +23,8 @@ func locAt(ts time.Time, lat, lon, hdop float64) map[string]any { "value": map[string]any{"latitude": lat, "longitude": lon, "hdop": hdop}} } -// V1 — a location signal has value_number NULL and loc_* set; the fold must keep -// value_number NULL and carry loc columns exactly like the recompute. +// V1 — a location signal has value_number NULL and loc_* set; the recompute must keep +// value_number NULL and carry the loc columns of the newest fix. func TestVerify01_LocationSignalNullValueNumber(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -35,7 +36,7 @@ func TestVerify01_LocationSignalNullValueNumber(t *testing.T) { drainNoFlush(t, ctx, runner) seedRawStatus(t, db, "v1b", subj, base.Add(2*time.Minute), locAt(base.Add(2*time.Minute), 42.3, -83.1, 0.9)) drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) // the real contract: value_number (whatever it is) matches the recompute + require.NoError(t, mat.RecomputeRollup(ctx)) got := dumpRollupMap(t, ctx, db)[subj+"|currentLocationCoordinates"] assert.InDelta(t, 42.3, got.locLat, 1e-9, "loc_lat folds to the newest fix") assert.True(t, got.locTS.Equal(base.Add(2*time.Minute)), "loc_ts is the newest fix") @@ -56,13 +57,13 @@ func TestVerify02_IntermittentLocation(t *testing.T) { drainNoFlush(t, ctx, runner) seedRawStatus(t, db, "v2old", subj, base.Add(1*time.Minute), locAt(base.Add(1*time.Minute), 1.0, 2.0, 9.0)) // older loc drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) + require.NoError(t, mat.RecomputeRollup(ctx)) loc := dumpRollupMap(t, ctx, db)[subj+"|currentLocationCoordinates"] assert.True(t, loc.locTS.Equal(base.Add(5*time.Minute)), "loc_ts stays the newest fix despite a later-arriving older one") assert.InDelta(t, 40.0, loc.locLat, 1e-9) } -// V3 — a fat single snapshot with many (subject,name) pairs folds exactly in one window. +// V3 — a fat single snapshot with many (subject,name) pairs decodes exactly in one window. func TestVerify03_ManyKeysOneSnapshot(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -85,7 +86,7 @@ func TestVerify03_ManyKeysOneSnapshot(t *testing.T) { } require.NoError(t, tx.Commit()) drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) + require.NoError(t, mat.RecomputeRollup(ctx)) assert.Len(t, dumpRollupMap(t, ctx, db), 12, "6 subjects x 2 names = 12 rollup rows") } @@ -104,13 +105,13 @@ func TestVerify04_ByteBudgetPagination(t *testing.T) { }) seedRawStatusOneSnapshot(t, db, subj, base, 1100) // > windowReadChunk so the byte path paginates drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) + require.NoError(t, mat.RecomputeRollup(ctx)) assert.EqualValues(t, 1100, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) assert.Positive(t, intermediate, "the byte-budget path split the span into multiple windows") } // V5 — crash at the LAST intermediate window (most windows durable, cursor not advanced); -// restart converges exactly for BOTH base rows and the rollup. +// restart converges to exactly-once base rows (no row lost, none double-counted). func TestVerify05_CrashLastIntermediateWindow(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -133,8 +134,8 @@ func TestVerify05_CrashLastIntermediateWindow(t *testing.T) { assert.EqualValues(t, 0, readCursor(t, ctx, db), "cursor not advanced on a partial span") runner2, mat2 := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(2) }) drainNoFlush(t, ctx, runner2) + require.NoError(t, mat2.RecomputeRollup(ctx)) assert.EqualValues(t, 9, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) - assertMatchesRecompute(t, ctx, db, mat2) } // V6 — a paginated span mixes decodable rows with rows that decode to NOTHING (wrong-chain @@ -164,12 +165,12 @@ func TestVerify06_MixedNonDecodableRows(t *testing.T) { runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(3) }) drainNoFlush(t, ctx, runner) assert.Positive(t, readCursor(t, ctx, db), "cursor advanced past non-decodable rows") + require.NoError(t, mat.RecomputeRollup(ctx)) assert.EqualValues(t, 6, dumpRollupMap(t, ctx, db)[good+"|speed"].count, "only the 6 vehicle rows counted") - assertMatchesRecompute(t, ctx, db, mat) } // V7 — an out-of-order batch 100 days older than the existing latest: recency unchanged, -// count increments, first_seen moves back, exactly matching the recompute. +// count increments, first_seen moves back in the recomputed rollup. func TestVerify07_DeepOutOfOrder(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -182,15 +183,15 @@ func TestVerify07_DeepOutOfOrder(t *testing.T) { old := now.AddDate(0, 0, -100) seedRawStatus(t, db, "v7old", subj, old, speedAt(old, 12)) drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) + require.NoError(t, mat.RecomputeRollup(ctx)) got := dumpRollupMap(t, ctx, db)[subj+"|speed"] assert.EqualValues(t, 2, got.count) assert.EqualValues(t, 70, got.valueNumber.Float64, "recency unchanged by the 100-day-old arrival") assert.True(t, got.firstSeen.Equal(old), "first_seen moved back 100 days") } -// V8 — interleave a full RecomputeRollup with incremental folds: the incremental path must -// stay exact when it continues on top of a freshly recomputed rollup (self-healing). +// V8 — RecomputeRollup mid-stream then again after more data: rebuilding on top of a +// previously recomputed rollup must stay exact (the DELETE+INSERT rebuild is idempotent). func TestVerify08_RecomputeInterleave(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -204,6 +205,6 @@ func TestVerify08_RecomputeInterleave(t *testing.T) { require.NoError(t, mat.RecomputeRollup(ctx)) // rebuild from base mid-stream seedRawStatus(t, db, "v8c", subj, base.Add(3*time.Minute), speedAt(base.Add(3*time.Minute), 30)) drainNoFlush(t, ctx, runner) - assertMatchesRecompute(t, ctx, db, mat) + require.NoError(t, mat.RecomputeRollup(ctx)) assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) } diff --git a/tests/latest_kv_extended_test.go b/tests/latest_kv_extended_test.go index 40d3490..ef6e6c4 100644 --- a/tests/latest_kv_extended_test.go +++ b/tests/latest_kv_extended_test.go @@ -37,9 +37,13 @@ func TestLatestKVExt_AllLatestAndAvailableServeMatchRollup(t *testing.T) { mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) require.NoError(t, err) mat.WithLatestPublisher(app.NewLatestKVPublisher(store, nil, zerolog.Nop())) + mat.WithDailyRollup(materializer.DailyRollupOn, 0) runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 3, drainRunner(t, ctx, runner)) + // All readings are settled (< today's boundary), so the daily refresh + // leaves the rollup complete for this subject — the comparison baseline. + require.NoError(t, mat.RunDailyRollupRefresh(ctx, time.Now().UTC().Truncate(24*time.Hour))) rollupQ := duck.NewLakeQueries(svc) serveQ := duck.NewLakeQueries(svc). @@ -112,11 +116,11 @@ func TestLatestKVExt_AllLatestAndAvailableServeMatchRollup(t *testing.T) { } // TestSignalSummaries_DailyServingUnionExact pins the (rollup ∪ tail) union: -// against a day-stale rollup (simulated by swapping in the shadow table the -// daily refresh maintains), summaries under LAKE_ROLLUP_DAILY_SERVING must -// equal the answers the fresh rollup gave — counts, first/last seen, and a -// name first seen only AFTER the watermark (which exists in no rollup row at -// all and only the tail can count). +// against the genuinely day-stale rollup mode on produces (exact as of the +// watermark, blind to the tail), summaries under LAKE_ROLLUP_DAILY_SERVING +// must equal the answers a full-recompute oracle gives — counts, first/last +// seen, and a name first seen only AFTER the watermark (which exists in no +// rollup row at all and only the tail can count). func TestSignalSummaries_DailyServingUnionExact(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -126,7 +130,7 @@ func TestSignalSummaries_DailyServingUnionExact(t *testing.T) { watermark := day.AddDate(0, 0, 2) runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) + m.WithDailyRollup(materializer.DailyRollupOn, 0) }) // Pre-watermark history: two names, multiple readings, a redelivery. seedRawStatus(t, db, "su-1", subject, day.Add(1*time.Hour), speedAt(day.Add(1*time.Hour), 10)) @@ -135,32 +139,31 @@ func TestSignalSummaries_DailyServingUnionExact(t *testing.T) { drainNoFlush(t, ctx, runner) require.NoError(t, mat.RunDailyRollupRefresh(ctx, watermark)) - // Post-watermark tail: more speed readings and a name BORN after W. + // Post-watermark tail: more speed readings and a name BORN after W. The + // rollup stays exact-as-of-W (nothing folds the tail until the next + // boundary) — the day-stale state daily serving exists for. newTS := watermark.Add(30 * time.Minute) seedRawStatus(t, db, "su-3", subject, newTS, speedAt(newTS, 30), map[string]any{"name": "powertrainRange", "timestamp": newTS.Format(time.RFC3339Nano), "value": 250.0}) drainNoFlush(t, ctx, runner) - // Expected: the answers off the FRESH per-pass rollup (exact today). - expected, err := duck.NewLakeQueries(svc).GetSignalSummaries(ctx, subject, nil) + // Sanity: the plain rollup read is WRONG (missing the tail-born name) — + // the union assertion below is not vacuous. + stale, err := duck.NewLakeQueries(svc).GetSignalSummaries(ctx, subject, nil) require.NoError(t, err) - require.NotEmpty(t, expected) + require.NotEmpty(t, stale) - // Simulate the step-4 flip: signals_latest becomes the daily table's - // content — exact as of the watermark, blind to the tail. - _, err = db.ExecContext(ctx, "DELETE FROM lake.signals_latest") - require.NoError(t, err) - _, err = db.ExecContext(ctx, "INSERT INTO lake.signals_latest SELECT * FROM lake.signals_latest_daily") + got, err := duck.NewLakeQueries(svc).WithDailyServingRollup(true).GetSignalSummaries(ctx, subject, nil) require.NoError(t, err) - // Sanity: the plain rollup read is now WRONG (missing the tail) — the - // union assertion below is not vacuous. - stale, err := duck.NewLakeQueries(svc).GetSignalSummaries(ctx, subject, nil) + // Expected: the answers off a full-recompute oracle (rebuilds + // lake.signals_latest in place, so it runs AFTER the union was captured). + oracleRecompute(t, ctx, db) + expected, err := duck.NewLakeQueries(svc).GetSignalSummaries(ctx, subject, nil) require.NoError(t, err) + require.NotEmpty(t, expected) require.NotEqual(t, len(expected), len(stale), "day-stale rollup must be missing the tail-born name") - got, err := duck.NewLakeQueries(svc).WithDailyServingRollup(true).GetSignalSummaries(ctx, subject, nil) - require.NoError(t, err) require.Len(t, got, len(expected)) for i := range expected { assert.Equal(t, expected[i].Name, got[i].Name) diff --git a/tests/latest_kv_negative_test.go b/tests/latest_kv_negative_test.go index c58cdef..5eff46e 100644 --- a/tests/latest_kv_negative_test.go +++ b/tests/latest_kv_negative_test.go @@ -195,6 +195,7 @@ func TestLatestKVNegative_ShadowServesRollupResult(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 1, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) shadowQ := negativeQueries(t, svc, store, duck.KVNegativeShadow) rollupQ := duck.NewLakeQueries(svc) diff --git a/tests/latest_kv_read_test.go b/tests/latest_kv_read_test.go index 9cc6581..7eb986d 100644 --- a/tests/latest_kv_read_test.go +++ b/tests/latest_kv_read_test.go @@ -48,6 +48,7 @@ func TestLatestKV_ServeMatchesRollupAndFallsBack(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 2, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) rollupQ := duck.NewLakeQueries(svc) serveQ := duck.NewLakeQueries(svc).WithLatestKV(store, duck.KVReadServe, zerolog.Nop()) @@ -106,6 +107,7 @@ func TestLatestKV_ShadowServesRollupResult(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 1, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) shadowQ := duck.NewLakeQueries(svc).WithLatestKV(store, duck.KVReadShadow, zerolog.Nop()) got, err := shadowQ.GetLatestSignals(ctx, subject, latestArgsFor("speed")) diff --git a/tests/latest_kv_test.go b/tests/latest_kv_test.go index 6bfed1c..e15b4e7 100644 --- a/tests/latest_kv_test.go +++ b/tests/latest_kv_test.go @@ -66,6 +66,7 @@ func TestLatestKV_PublishedAtDecode_MatchesRollup(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 2, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) entry, err := store.GetEntry(ctx, subject) require.NoError(t, err) @@ -109,6 +110,7 @@ func TestLatestKV_BootstrapFromRollup(t *testing.T) { runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). WithDuckLake(mat) require.Equal(t, 2, drainRunner(t, ctx, runner)) + refreshRollup(t, ctx, mat, day.AddDate(0, 0, 1)) store := newLatestKVStore(t, "boot") require.NoError(t, store.BootstrapFromRollup(ctx, db, false)) diff --git a/tests/latest_kv_watermark_test.go b/tests/latest_kv_watermark_test.go index d2d12c8..84772b3 100644 --- a/tests/latest_kv_watermark_test.go +++ b/tests/latest_kv_watermark_test.go @@ -25,9 +25,9 @@ import ( // TestLatestKV_ReconcileReplaysTailSinceWatermark manufactures the exact hole // the tail replay closes: subject B's first-ever reading lands AFTER the // watermark, its publish is "lost" (the bucket never saw it), and the rollup -// is day-stale (B absent — simulated by deleting B's rollup rows, which is -// what a daily-refreshed rollup looks like before its next fold). The -// reconcile must still produce B's key, or negative serving would lie. +// is day-stale (B absent — under mode on the daily refresh really does leave +// post-watermark subjects invisible until the next boundary). The reconcile +// must still produce B's key, or negative serving would lie. func TestLatestKV_ReconcileReplaysTailSinceWatermark(t *testing.T) { ctx := context.Background() svc := newLakeService(t, t.TempDir()) @@ -43,11 +43,12 @@ func TestLatestKV_ReconcileReplaysTailSinceWatermark(t *testing.T) { // No KV publisher wired: every "publish" is lost, the bucket starts empty. runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) + m.WithDailyRollup(materializer.DailyRollupOn, 0) }) seedRawStatus(t, db, "wmA", subjA, oldTS, speedAt(oldTS, 33)) drainNoFlush(t, ctx, runner) - // The real writer establishes the watermark row latestkv must find. + // The real writer establishes the watermark row latestkv must find, and + // seeds the rollup (which covers A: oldTS < watermark). require.NoError(t, mat.RunDailyRollupRefresh(ctx, watermark)) // B is born after the watermark, with a same-timestamp collision (the @@ -59,11 +60,9 @@ func TestLatestKV_ReconcileReplaysTailSinceWatermark(t *testing.T) { seedRawStatus(t, db, "wmB-3", subjB, locTS, locFixAt(locTS, 42.33, -83.05)) drainNoFlush(t, ctx, runner) - // Simulate the post-flip world: the rollup is day-stale and has never seen - // B. (Pre-flip the per-pass fold keeps it fresh, so delete B's rows.) - _, err := db.ExecContext(ctx, "DELETE FROM lake.signals_latest WHERE subject = ?", subjB) - require.NoError(t, err) - + // The rollup is now GENUINELY day-stale: under the daily refresh nothing + // folds B's post-watermark rows until the next boundary, so B is absent + // from lake.signals_latest — exactly the hole the tail replay must cover. require.NoError(t, store.ReconcileFromRollup(ctx, db)) // A: restored from the rollup pass, as before. @@ -107,16 +106,17 @@ func TestLatestKV_BootstrapReplaysTail(t *testing.T) { day := time.Now().UTC().AddDate(0, 0, -2).Truncate(24 * time.Hour) watermark := day.AddDate(0, 0, 1) runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { - m.WithDailyRollup(materializer.DailyRollupShadow, 0) + m.WithDailyRollup(materializer.DailyRollupOn, 0) }) // Establish the watermark first, over an empty pre-boundary lake. require.NoError(t, mat.RunDailyRollupRefresh(ctx, watermark)) + // The subject is born after the watermark: under the daily refresh the + // rollup never sees it until the next boundary — the day-stale hole the + // bootstrap's tail replay must cover. newTS := watermark.Add(45 * time.Minute) seedRawStatus(t, db, "wmboot-1", subj, newTS, speedAt(newTS, 88)) drainNoFlush(t, ctx, runner) - _, err := db.ExecContext(ctx, "DELETE FROM lake.signals_latest WHERE subject = ?", subj) - require.NoError(t, err) require.NoError(t, store.BootstrapFromRollup(ctx, db, false)) entry, err := store.GetEntry(ctx, subj) diff --git a/tests/rollup_helpers_test.go b/tests/rollup_helpers_test.go new file mode 100644 index 0000000..6aa5441 --- /dev/null +++ b/tests/rollup_helpers_test.go @@ -0,0 +1,144 @@ +// rollup_helpers_test.go — shared fixtures for the lake.signals_latest rollup +// tests: the row shape + dump, the materializer/runner constructor, the +// no-flush drain, and the full-recompute oracle. Extracted from the (deleted) +// incremental-fold differential test when dq#55 step 5 removed the per-pass +// fold; the daily refresh (daily_rollup.go) is now the rollup's only +// steady-state writer, and these helpers are how its output is checked. +package tests + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/DIMO-Network/dq/internal/materializer" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type rollupRow struct { + subject, name string + bucket int + timestamp time.Time + valueNumber sql.NullFloat64 + valueString sql.NullString + locLat, locLon float64 + locHdop, locHeading float64 + locTS time.Time + count int64 + firstSeen, lastSeen time.Time +} + +func dumpRollupMap(t *testing.T, ctx context.Context, db *sql.DB) map[string]rollupRow { + t.Helper() + rows, err := db.QueryContext(ctx, + `SELECT subject, name, subject_bucket, "timestamp", value_number, value_string, + loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen + FROM lake.signals_latest ORDER BY subject, name`) + require.NoError(t, err) + defer rows.Close() //nolint:errcheck + out := map[string]rollupRow{} + for rows.Next() { + var r rollupRow + require.NoError(t, rows.Scan(&r.subject, &r.name, &r.bucket, &r.timestamp, &r.valueNumber, &r.valueString, + &r.locLat, &r.locLon, &r.locHdop, &r.locHeading, &r.locTS, &r.count, &r.firstSeen, &r.lastSeen)) + out[r.subject+"|"+r.name] = r + } + require.NoError(t, rows.Err()) + return out +} + +func incrRunner(t *testing.T, ctx context.Context, db *sql.DB, opts ...func(*materializer.DuckLakeMaterializer)) (*materializer.Runner, *materializer.DuckLakeMaterializer) { + t.Helper() + mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + for _, o := range opts { + o(mat) + } + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) + return runner, mat +} + +// drainNoFlush runs the decode loop to completion WITHOUT calling FlushRollup. +// Since dq#55 step 5 nothing maintains lake.signals_latest at decode time, so +// after this drain the rollup holds whatever the last daily refresh (or +// recompute) left — tests asserting rollup content must run +// RunDailyRollupRefresh (mode on) or RecomputeRollup explicitly first. +func drainNoFlush(t *testing.T, ctx context.Context, r *materializer.Runner) { + t.Helper() + for { + n, err := r.RunOnce(ctx) + require.NoError(t, err) + if n == 0 { + return + } + } +} + +// oracleRecompute rebuilds lake.signals_latest IN PLACE from the full base +// with a fresh default-mode materializer (RecomputeRollup is deliberately +// disabled under mode on) and returns the rows — the exactness oracle for the +// daily-refresh tests, valid when every seeded row is stamped before the last +// refreshed boundary. Capture the refresh's answer BEFORE calling this: the +// rebuild overwrites the table being checked. +func oracleRecompute(t *testing.T, ctx context.Context, db *sql.DB) map[string]rollupRow { + t.Helper() + oracle, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + require.NoError(t, oracle.RecomputeRollup(ctx)) + return dumpRollupMap(t, ctx, db) +} + +// assertRollupMatchesOracle asserts the current lake.signals_latest (as the +// daily refresh left it) is column-for-column identical to a full recompute +// over the deduped base. Valid whenever every seeded row's timestamp is before +// the last refreshed boundary (all data settled), which callers arrange. +func assertRollupMatchesOracle(t *testing.T, ctx context.Context, db *sql.DB) { + t.Helper() + got := dumpRollupMap(t, ctx, db) + want := oracleRecompute(t, ctx, db) + keys := map[string]bool{} + for k := range got { + keys[k] = true + } + for k := range want { + keys[k] = true + } + for k := range keys { + g, okG := got[k] + w, okW := want[k] + require.Truef(t, okG, "%s present after the oracle recompute but MISSING from the daily-refreshed rollup", k) + require.Truef(t, okW, "%s present in the daily-refreshed rollup but MISSING after the oracle recompute", k) + assert.Equalf(t, w.count, g.count, "%s count", k) + assert.Truef(t, w.timestamp.Equal(g.timestamp), "%s timestamp: oracle=%s refresh=%s", k, w.timestamp, g.timestamp) + assert.Equalf(t, w.valueNumber, g.valueNumber, "%s value_number", k) + assert.Equalf(t, w.valueString, g.valueString, "%s value_string", k) + assert.Truef(t, w.firstSeen.Equal(g.firstSeen), "%s first_seen: oracle=%s refresh=%s", k, w.firstSeen, g.firstSeen) + assert.Truef(t, w.lastSeen.Equal(g.lastSeen), "%s last_seen: oracle=%s refresh=%s", k, w.lastSeen, g.lastSeen) + assert.InDeltaf(t, w.locLat, g.locLat, 1e-9, "%s loc_lat", k) + assert.InDeltaf(t, w.locLon, g.locLon, 1e-9, "%s loc_lon", k) + assert.InDeltaf(t, w.locHdop, g.locHdop, 1e-9, "%s loc_hdop", k) + assert.Truef(t, w.locTS.Equal(g.locTS), "%s loc_ts: oracle=%s refresh=%s", k, w.locTS, g.locTS) + } +} + +// refreshRollup makes lake.signals_latest current through boundary via the +// PRODUCTION maintenance path: it enables the daily rollup on mat and runs one +// refresh. Since dq#55 step 5 nothing maintains the rollup at decode time, so +// any test that drains and then reads rollup-served answers calls this first. +// boundary must be a UTC midnight covering every seeded timestamp; readings +// stamped before an already-set watermark are healed via the late-subject path +// (they were marked during the drain), so successive drain+refresh phases with +// increasing boundaries stay exact. Note mat is mode-on afterwards +// (RecomputeRollup on it is refused — use oracleRecompute for a DR rebuild). +func refreshRollup(t *testing.T, ctx context.Context, mat *materializer.DuckLakeMaterializer, boundary time.Time) { + t.Helper() + mat.WithDailyRollup(materializer.DailyRollupOn, 0) + require.NoError(t, mat.RunDailyRollupRefresh(ctx, boundary)) +} + +func odoAt(ts time.Time, v float64) map[string]any { + return map[string]any{"name": "powertrainTransmissionTravelledDistance", "timestamp": ts.Format(time.RFC3339Nano), "value": v} +}