feat: DuckLake parse-on-read — materializer, query backend, consumer floor (dq side of din#2)#7
Merged
Merged
Conversation
Per-connection bootstrap (memory_limit/threads/spill/object_cache, httpfs+aws+secret when S3 enabled), explicit per-day read_parquet glob builders for raw and decoded datasets, shared HashBucket contract for latest/summary buckets, and pre-baked duckdb extensions in the image.
Decodes raw cloudevents via model-garage (vendor modules wired: AutoPi/Ruptela/Kaufmann/HashDog + default-module fallback with partial salvage), ports dis pruning + coordinate merging, writes sorted/zstd/ bloom signal+event parquet, latest/summary buckets (lastSeen virtual row, (0,0) location exclusion, footer batch-stamp idempotency), and a crash-safe manifest+watermark commit protocol with crash-injection test matrix.
ch.Service-compatible Queries over decoded parquet: epoch-math interval bucketing (CH toStartOfInterval parity), FILTER-clause aggregations, arg_min/arg_max FIRST/LAST, mode/string_agg string aggs, latest buckets with lastSeen virtual row and (0,0)-exclusion via *_nonzero columns, list_has_any tag filters, glob() pre-expansion for missing partitions. Location polygon/circle filters TODO (spatial extension).
…ring Backend interface split (signal/latest/summary/events vs CH-only segments); duck.Queries satisfies it with compile-time assertions. Shadow mode serves ClickHouse, async-compares DuckDB results (epsilon floats, bounded concurrency, panic-safe) into dq_shadow_mismatch_total. MATERIALIZER_ENABLED starts the post-fact decode loop with vendor modules from new chain settings.
Replaces the ClickHouse cloud_event index + SeekToRow path: list/latest/ available-types queries scan raw/type=/date= partitions directly with data inline, dedup on header key, week-chunked newest-first walk for latest, glob() pre-expansion for missing partitions. Tombstone voiding parity is a documented TODO. eventrepo facade (grpc.SearchOptions -> RawFilter) is the follow-up wiring step.
Reader used bucket=%d while the materializer writes bucket=%03d — every latest/summary lookup for buckets under 100 silently returned empty. Regression test pins the padded format.
string_agg(DISTINCT) without ORDER BY is insertion-ordered, so shadow comparisons against ClickHouse fired false mismatches on every UNIQUE aggregation with 2+ values.
Upstream model-garage init() registers tesla.Module for TeslaSource; the local registry mirror omitted it, so Tesla raw status fell through to the default module — decoded tables silently missing an entire oracle while live decodestream decoded it fine.
Within-batch event dedup on header Key() keeps at-least-once ingest duplicates out of decoded rows and summary counts; dimo.status signal decoding gated to vehicle-NFT subjects (dis/din parity — decoded tables previously gained rows the ClickHouse path never had).
Writes raw cloudevent bundles byte-identical to din's sink output (hive layout, ingest- naming, sorted/zstd/bloom), runs the materializer post-fact, then asserts DuckDB results: hand-computed aggregation with duplicate-event collapse, latest + lastSeen, available signals, the vehicle gate (raw queryable, decoded excluded), verbatim raw payloads, watermark publication, and incremental freshness on a follow-up poll.
The verbatim fullInputJSON device payload from model-garage's ruptela conversion tests (the same payload dis's integration suite posts over mTLS) flows raw-parquet -> materializer -> DuckDB, then a real gqlgen execution of signalsLatest must return dis's hand-verified values bit-for-bit: speed 31.24609375, powertrainType COMBUSTION, oil LOW, tire pressure 262.00088, fuel 19.200000000000003, GPS 52.2721466 / -0.9014316 hdop 0.6, plus lastSeen and per-signal timestamps.
One-vehicle 7-day hourly aggregation over 200 vehicles x 30 days x 3 signals/minute: 8ms cold, 6ms warm, 15ms full month (local NVMe; S3 adds GET latency). Gated behind -perf flag; plan targets were <1s warm / <3s cold — beaten by two orders of magnitude on engine cost.
- shadow: compare []*vss.Signal as canonically sorted copies; neither backend orders GROUP BY output, so positional diff fired dq_shadow_mismatch_total on nearly every latest-signals call - duck: ORDER BY name on latest queries for deterministic results - duck: wrap LatestCloudEvent miss in ErrNotFound so the eventrepo facade can tell absence from I/O failure - materializer: synthesized location signal inherits header from the active triple, not signals[0], when a batch mixes sources
- raw_types_test: every cloudevent type constant through duck.Raw list/latest/available plus source/producer/id filters, tombstone voids_id, data_base64; new-constant drift guard - segments_graphql_test: all six detection mechanisms + dailyActivity via real gqlgen execution over a fake SegmentsBackend with real DuckDB enrichment; validation, config plumb-through, limit/after, timezone handling - duck: SET TimeZone='UTC' per connection — naive make_timestamp window literals resolved in host zone, shifting every sub-day window query on non-UTC hosts (found by the new tests) - duck: scan voids_id now that cloudevent writes it
startMaterializer now accepts path-like parquet buckets and runs over the local filesystem instead of rejecting them; duck already handled local globs. Publishes are atomic (temp + fsync + rename) so din's compactor never reads a torn watermark. - internal/fsstore: production materializer.ObjectStore over a root dir; missing prefixes list empty (poll-before-first-write), sorted List, temp files hidden, ErrNotFound wrapped - tests: pipeline/parity/segments suites now run on the production store; test-only fsStore deleted - README: storage backend table + single-node quickstart
-perf gated. Raw din-layout bundles through one materializer pass; reports events/s and signals/s, gates at >1k events/s.
…g metrics First mass-scale roadmap item: the single-replica materializer wall. - materializer: per-batch fan-out (Workers, default GOMAXPROCS) across raw fetch+decode, model-garage conversion, data-object writes, and bucket read-merge-writes; output stays deterministic (sequential dedup, writers re-sort). 29k -> 39k events/s on NVMe; bigger win on S3 where the fan-out hides GET/PUT latency - decoded-layer compaction: merges closed date partitions into one file each (CompactInterval/CompactMinFiles). Aggregations don't dedup decoded rows, so the protocol never lets sources and target be visible together: stage outside query globs -> manifest -> delete sources -> publish. Crash matrix proves recovery converges from every interruption point; staged-object existence is the recovery discriminator (target-exists would lose late rows on recompaction crashes) - metrics: dq_materializer_lag_seconds (decode-lag SLO), batches/ rows/errors/compactions counters - fsstore + s3ObjectStore gain DeleteObject (CompactStore) - e2e: query results proven identical across compaction
- materializer: BatchMaxBytes (1GiB) cuts batches by aggregate raw size — the whole batch decodes in memory, file count alone let 64x128MB batches OOM; decoded[i] released as the dedup pass consumes it - compaction: 2GiB partition size cap (skip+warn until chunked merge); recompaction threshold drops to 2 once compacted.parquet exists so late batches can't sit uncompacted below CompactMinFiles forever; S3 overwrite-visibility note on compactedName - lag gauge reset unconditionally after a drained type pass (was guarded on len(batches)>0 — stale forever once caught up) - latest+summary bucket updates run concurrently (disjoint key sets) - s3 DeleteObject treats NoSuchKey as done (MinIO; recovery idempotence)
docker/docker alerts stay open: the patched v29.3.1 is not published to the docker/docker Go module path, and the dependency is test-only surface (testcontainers client for ClickHouse integration tests).
- fsstore: fsync parent directory after rename (watermark.json must never silently vanish on crash) - duck: LatestCloudEvent scans the floor day on zero-width windows (loop boundary was exclusive) - materializer settings wired: MATERIALIZER_WORKERS/BATCH_FILES/ BATCH_BYTES, COMPACT_INTERVAL_SECONDS, COMPACT_MIN_FILES
- internal/audit: store-agnostic invariant checker (no duplicate decoded rows, decoded ⊆ raw, watermark cursor shape, no leftover compaction manifests) — built for chaos tests now, runnable against the production bucket before cutover - tests/chaos: SIGKILLs the materializer process 15x at random points under live seeding, drains, then proves every seeded event decoded exactly once with zero violations (DQ_CHAOS=1 gated) - tests/minio: full pipeline (s3 materializer store, DuckDB httpfs CREATE SECRET + s3:// globs, compaction visibility) against a real S3 API via local MinIO (skips when minio absent)
Scale-out for the two single-writer components, plus the deploy artifacts that enforce single-writer until then. - materializer sharding (MATERIALIZER_SHARD_INDEX/COUNT): shards own disjoint raw partitions by partition hash, write per-shard watermark files (_state/watermark-pNNNofMMM.json) and per-shard latest/summary namespaces (latest/shard=N/bucket=M) — no two shards ever write the same object. Decoded compaction shards by the same hash. - duck: latest/summary reads glob both the single-replica and sharded namespaces; max/arg_max/sum aggregations merge shard files natively, so mixed layouts and migrations read correctly. Proven end-to-end: two shards, global latest correct, summary counts exactly-once, audit clean (sharding_test.go) - settle window (90s default): the cursor never advances over a raw key younger than the slowest plausible sink flush — closes a window where a slow PUT could land below the cursor and never decode - audit: merges sharded watermark files; flags split-brain (two cursors claiming one partition) - charts/dq: standard chart with strategy=Recreate (single-writer guard during deploys) and PrometheusRule alerts for decode lag, stalled watermark, and shadow mismatches
Foundation + materializer write path for the DuckLake migration; the bucket/query paths are untouched, so this is additive behind config. - duck: DUCKLAKE_ENABLED attaches a DuckLake catalog as schema 'lake' per connection (postgres extension when the DSN is Postgres, else a file catalog for tests/single-node); reuses extension_directory + S3 secret handling - materializer DuckLakeWriter: one catalog txn inserts the decoded batch and advances a per-partition cursor (lake.ingest_progress) — the transaction is the manifest+watermark, exactly-once by construction. Concurrent same-partition races abort at commit (no double-insert), proven in a phase-0 spike. Reuses decodeBatch + pendingBatches (settle window + ownership) verbatim - projects the catalog cursor to decoded/v1/_state/watermark.json so din's raw compactor needs no catalog access - e2e (file catalog): rows land in lake.signals, cursor advances, watermark projected, re-run is exactly-once Remaining: query Backend over lake.signals/events (phase 3), wiring + shadow parity (phase 4), Postgres-gated concurrency test, deploy + bucket-path retirement (phase 5).
din's DuckLake PR standardized on github.com/duckdb/duckdb-go/v2; dq and din attach the same catalog, so they must share a driver/DuckDB version. Identical NewConnector(dsn, connInitFn) signature made it a drop-in swap across duck.go, the installext prebake tool (now also bakes ducklake), and the tools pin. dq is now a CGO build like din. Full suite + DuckLake e2e + chaos green on the new driver.
Converges the dq decoded materializer onto din's DuckLake raw layer (PR #2). Replaces the S3-hive read path with an incremental snapshot diff over the shared catalog's lake.raw_events: - DuckLakeMaterializer: per pass, read ducklake_table_changes(..., cursor+1, head) inserts, decode (model-garage), write lake.signals/ events, advance a single snapshot-id cursor in lake.ingest_progress — all in one transaction. Exactly-once by construction; concurrent decoders conflict at commit and retry from the new cursor. - No S3 LIST, no watermark.json, no settle window: a row is visible only after din committed its snapshot, so there is no pre-PUT race. Coordination with din is LAKE_SNAPSHOT_RETENTION (cursor must stay in the window), not a watermark file. - decodeEvents: per-event type routing (status->signals, events-> events) over a reconstructed delta, reusing convertSignals/Events. - e2e (file catalog): seed raw_events as din's sink does -> decode -> lake.signals, exactly-once on re-run, incremental on new snapshot. Retires the prior bucket-write DuckLakeWriter. Bucket query/retire and QUERY_BACKEND=ducklake wiring follow.
whereClauseQ emitted time >= ? for the After filter. ClickHouse eventrepo uses strict time > ?, so events at exactly the After timestamp were incorrectly included in lake results. Change to >. The hive Raw path shares whereClauseQ and is being retired; CH parity takes precedence. Test added: TestAfterBoundaryIsStrict — asserts an event at exactly the After timestamp is excluded while an event 1ms later is included.
filterFromAdvanced now mirrors eventrepo.AdvancedSearchOptionsToQueryMod field-for-field. RawFilter and whereClauseQ are extended to match: - Subject: multi-value IN (Subjects []string) instead of In[0] only. - String fields (Type/Source/Producer/ID/DataVersion/Extras): NotIn (NOT IN) support alongside the existing In. - Tags: ContainsAll (list_has_all), NotContainsAny (NOT list_has_any), NotContainsAll (NOT list_has_all) — all over extras.tags as VARCHAR[]. - Extras: IN / NOT IN on the raw extras text column. - Or clauses: return errOrClauseUnsupported (not silently over-return). Tests added: TestMultiSubjectIN, TestStringNotIn, TestTagsContainsAll, TestTagsNotContainsAny, TestExtrasFilter, TestOrClauseReturnsError.
CreateGRPCServer now returns (*grpc.Server, func(), error). The cleanup func releases the DuckDB connection and catalog attach after the server stops serving. main.go defers it after runnerGroup.Wait() so it runs only after GracefulStop completes. Previously cleanup was dropped on the success path, leaking the duck.Service for the process lifetime.
Replace the dead duckSvc == nil guard in the shadow arm of newEventService with settings.DuckLakeCatalogDSN == "". duckSvc is never nil for any non-clickhouse backend (newQueryBackend always allocates a duck.Service), so the old guard never fired. Without a catalog DSN the duck.Service has no lake catalog attached, and constructing a LakeEventService over it would fail at query time on lake.raw_events. The corrected guard falls back to a CH-only EventService when no DSN is configured, matching the same condition used in newQueryBackend to skip lake segment attachment.
Add white-box tests (package eventrepo) that use testutil.ToFloat64 to read dq_fetch_shadow_mismatch_total by method label: - TestShadowMismatchCounter_IncrementOnMismatch: counter goes up by 1 when primary and secondary return different results. - TestShadowMismatchCounter_NoIncrementOnMatch: counter stays flat when results are identical. Both tests call shadow.Wait() before reading the counter to ensure the shadow goroutine has finished. Deltas are taken against the per-label value before the call to avoid cross-test pollution from the shared global registry. The existing TestShadowEventService_MismatchIncrementsCounter is updated to reference the new companion test.
Add tests/fetch_lake_parity_test.go with a top-of-file parity coverage map (each CH eventrepo behaviour → where it is asserted in the suite) plus four new tests for the gaps that existing files did not cover: - TestBeforeBoundaryIsStrict: Before is exclusive upper bound (time < ?) - TestSourceINFilter: Source IN narrows results - TestProducerINFilter: Producer IN narrows results - TestTagsNotContainsAll: NOT list_has_all excludes events with all tags All eight parity bullets (ordering, strict After/Before, voiding, filter narrowing, all four tag operators, type summaries, dedup) are now demonstrably covered and cross-referenced. go build ./..., go test ./internal/... ./tests/ ./pkg/... -count=1, and golangci-lint run ./... all green. Only pre-existing pkg/eventrepo CH- container failures remain (TestGetLatestIndexKey, TestGetEventWithAllHeaderFields, TestGetData, TestListIndexesAdvanced — unchanged from branch base).
Add TimestampAsc bool to RawFilter, read opts.GetTimestampAsc().GetValue() in filterFromAdvanced (matching CH's nil/false→DESC, true→ASC decision), and thread the flag into queryLakeRaw's ORDER BY clause. GetLatestIndexAdvanced forces TimestampAsc=false before querying, mirroring CH's explicit override so "get latest" always returns the newest event regardless of the caller's flag. Two new tests assert ASC/DESC list order and that GetLatest remains correct with TimestampAsc=true.
Append a new section to the segments+fetch spec listing the 5 accepted divergences between QUERY_BACKEND=ducklake and ClickHouse that an operator must review before flipping the flag in production: Or-clause hard errors, empty-result OK vs NotFound, blob payload gap (flagged highest priority), >30-day ignition segment suppression, and app.New cleanup tidy-up.
Pre-cutover correctness + scale fixes on the DuckLake decoded layer, from the senior-review backlog (PLAN-ch-deprecation-issues.md). Also commits the already-validated Batch-1 fetch fixes (gRPC blob resolution, maxLakeQueryLimit clamp, ignition NULL filter). CHD-1 — partition + sort lake.signals / lake.events. Add a subject_bucket column (HashBucket(subject), stamped at decode time) and ALTER the tables to PARTITIONED BY (subject_bucket, day(timestamp)) + SORTED BY (subject, timestamp), mirroring din's raw_events. Per-vehicle reads add an inlined subject_bucket predicate (aggregations + latest/summary) so DuckLake prunes to one partition instead of scanning the fleet. Tables are created with the layout from the first write (decoded layer not yet in prod), so no re-materialization. CHD-2 — dedup every lake.signals read. At-least-once ingest can store the same (subject,name,timestamp) more than once with a different cloud_event_id; aggregations/latest/summary read the bare table and over-counted avg/count/sum/median. Centralize the segments-path QUALIFY ROW_NUMBER dedup (canonical = lowest cloud_event_id) so all callers inherit it. Collapsing also makes arg_max(value, timestamp) for latest unambiguous (tie-break). CHD-7 — idempotent decoded writes. The commit INSERT now anti-joins on the cloudevent identity (cloud_event_id, name, timestamp), pruned by subject_bucket, so a cloudevent redelivered in a later snapshot is not decoded and stored twice. CHD-8 — resolve blob payloads in the materializer. Payloads din externalizes to S3 (data_index_key under BlobKeyPrefix, no inline data) were discarded at decode, losing every >1 MiB payload. The materializer now downloads the blob before decoding, mirroring the fetch path's resolvePayload. Tests: local DuckLake file-catalog regression tests for each fix (blob decode, read dedup over-count, subject_bucket population, cross-snapshot idempotency). Build + lint clean; duck/materializer/tests suites green.
… sizing Observability and operability fixes for the DuckLake target mode, from the senior-review backlog (PLAN-ch-deprecation-issues.md). CHD-9 — single-writer materializer + idempotent bootstrap. The cursor row is now seeded once and every advance is a compare-and-swap UPDATE, so two concurrent first-writers can no longer both do a guard-less INSERT and double-decode the same range. (DuckLake has no PRIMARY KEY/UNIQUE, confirmed by probe, so the seed + CAS is the enforcement.) A chart render guard refuses MATERIALIZER_ENABLED=true with replicaCount>1 or autoscaling on — run decode as a dedicated 1-replica release, scale the query fleet separately. CHD-12 — instrument ducklake mode. The lake path emitted only cursor resets, so the DecodeLag/Stalled alerts (which watch dq_materializer_lag_seconds / batches_total) were dead in the target mode. It now emits decode lag, rows, batches, and cursor/head snapshot-id gauges, and records the skipped snapshot span on a cursor reset. New critical PrometheusRule on cursor resets (permanent skipped-data loss had no alert). CHD-13 — real readiness probe. /ready runs SELECT 1 FROM lake.signals LIMIT 0 (catalog reachable + extensions loaded + table present); the chart readiness probe now targets it on the serving port instead of a static mon-http 200, so a cold/catalog-down pod is pulled from the Service. Liveness stays static. CHD-20 — DuckDB runtime config + pod sizing. Set DUCKDB_MEMORY_LIMIT (~75% of pod, was unset → 80% of node RAM → OOM before spill), DUCKDB_THREADS (= CPU limit, was 64 throttled to 1), and DUCKDB_TEMP_DIRECTORY backed by a sized emptyDir spill volume. Prod query fleet sized for the ducklake target (4 CPU / 8 Gi) with HPA on CPU+memory. Tests: observeLakeLag gauge, ReadyHandler 503-on-failure, bootstrap seed-once. Build + lint clean; app/materializer/tests suites green; helm lint + template verified for dev and prod.
CHD-21 — Postgres catalog connection resilience. The DuckLake catalog is reached over a Postgres attach inside each DuckDB connection, but the pool set no SetConnMaxLifetime/SetConnMaxIdleTime, so a connection whose attach was poisoned by a PG blip stayed broken until pod restart. Add finite, configurable connection lifetime + idle-time (defaults 30m / 5m) so poisoned connections are recycled and re-bootstrap the attach. (HA Postgres itself — Patroni/RDS Multi-AZ + PgBouncer — is infrastructure, tracked separately.) CHD-22 — raise the gRPC message size limit. Cloudevent blob payloads run 4–50 MiB; the gRPC 4 MiB default silently truncated them once the fetch path started serving blobs. Set MaxSendMsgSize/MaxRecvMsgSize to 50 MiB. (Empty-list → NotFound parity and an authz interceptor need a client-contract / threat-model decision and are left as follow-ons.) Test: connection-recycling defaults are non-zero. Build + lint clean.
Both shadow validators (query and fetch) folded comparisons dropped under backpressure into the error counter, so a clean dq_shadow_mismatch_total could mean "the validator was saturated and didn't look", not "the lake matched". That makes the go/no-go cutover gate untrustworthy. Add dq_shadow_dropped_total / dq_fetch_shadow_dropped_total and increment those (not the error counters) when the concurrency semaphore is saturated. New DQShadowDropped warning so any dropped comparison surfaces — a clean gate now requires zero drops as well as zero mismatches. Test: a saturated shadow call increments the dropped counter and leaves the error counter untouched. Build + lint clean. (The reconciliation harness — bulk CH↔lake count/checksum sampling as an explicit pre-flip gate — is a larger piece left as a follow-on.)
A failed bootstrap statement was wrapped verbatim into the error, so CREATE SECRET (inlined S3 KEY_ID/SECRET) and ATTACH (Postgres DSN with password) leaked credentials into error messages and logs. Redact SECRET/ATTACH statements down to their kind before logging, mirroring din's lake.redact. Test: S3 keys and the Postgres password are masked; plain statements pass through. Build + lint clean.
Map empty List results to codes.NotFound, matching ClickHouse — the lake backend returns an empty slice with no error, which broke clients expecting NotFound. Reject client-supplied index keys containing path-traversal in ListCloudEventsFromIndex (defense-in-depth for the legacy CH/JSON fetch path, which dereferences the key; the lake path re-queries by (subject,id) and ignores it). Combined with the earlier MaxSend/RecvMsgSize bump this closes the gRPC contract items. Per-caller authz stays a network-policy/ClusterIP control (the FetchService is internal); a JWT interceptor is left opt-in to avoid breaking existing unauthenticated internal callers. Tests: empty list → NotFound; traversal key → InvalidArgument.
…CHD-34, CHD-33) CHD-34 — bound an open-ended lake fetch with a default 400-day lookback (CH eventrepo capped lookbacks; the lake path had none, so a filter without a time bound scanned all of raw_events). A point lookup by id stays unbounded so old events remain fetchable. CHD-33 — lock DuckDB aggregate semantics on the lake backend against ClickHouse golden vectors (median/empty-aggregate); the median↔median, mode↔topK, string_agg-DISTINCT↔groupUniqArray mapping is documented in aggregations.go and the bucket path exercises the same expressions in aggregations_test.go. Tests: list excludes a 500-day event while id-lookup still finds it; median golden vector; absent-signal aggregate is empty. (The materialized voiding column is a din-side write-time change — tracked there; the current NOT EXISTS self-join is correct.)
…riant (CHD-30/32/35) CHD-30 — extract the tombstone-voiding NOT EXISTS predicate into voidingClause() so the search (queryLakeRaw) and aggregate (type summaries) paths share one definition instead of two copies that can drift. CHD-32 — pin the decoded-table column contract with a test. The schema is otherwise implicit in the first materializer write (SignalRow/EventRow parquet template); a model-garage struct change would silently reshape the table and break appends. The test fails on drift; the column lists are the explicit, reviewed schema with a migration note. CHD-35 — document the hard UTC invariant at the bootstrap that enforces it: every pooled connection pins TimeZone='UTC' because raw_events."time" and the decoded timestamps are TIMESTAMP WITH TIME ZONE and queries inline naive make_timestamp literals. Build + lint clean; duck + tests suites green.
getLatest/Summary/availableSignals were a full-history GROUP BY per request (the migration dropped CH's precomputed latest tail). Add lake.signals_latest, a per-(subject,name) latest+summary rollup the materializer refreshes per batch (refreshRollup recomputes the touched subjects from the deduped base table in the commit transaction, so it advances atomically). The query layer serves GetAllLatestSignals / GetSignalSummaries / GetAvailableSignals from it when there is no source filter — O(distinct-names) instead of O(history). A source filter falls back to the full scan (still subject-pruned, CHD-1). The rollup is computed by the same max/arg_max + (0,0)-loc FILTER aggregation as getAllLatestSignalsLake, so it is a faithful materialized view (parity by construction). Partitioned by subject_bucket like the base table. Test: materialize three readings → latest/summary/available correct from the rollup; a fourth newer reading updates it incrementally. Full duck/materializer/ tests suites green; lint clean.
…D-14) The query path logged jwtSubject (the developer) but never the vehicle subject, so a "vehicle X is wrong" report was near un-root-causable. Add the asset DID (the vehicle being queried) to the request logger so every query-path log line is keyed by subject. Expose the decode backlog as a single number via a recording rule (dq:pipeline_snapshot_backlog = head - cursor snapshot id, using the gauges added in CHD-12); pair with din's head-age metric for end-to-end lag. Full OTLP distributed tracing (span export across GraphQL/gRPC→DuckDB and din→dq trace-context propagation) is the remaining infra piece; the subject-keyed logs + correlation deliver the root-cause-ability the finding called out.
…5, CHD-36) Add internal/reconcile: a bulk pre-flip gate that compares per-signal summaries (count, first/last seen) between the ClickHouse primary and the DuckLake secondary across a sample of vehicles, returning every (subject,name) disagreement. The migration previously had only organic shadow coverage; this is the explicit, exhaustive check to require green before flipping QUERY_BACKEND=ducklake. Add a `make test-gated` target so CI/operators run the gated suites (PG-concurrency, chaos, perf, MinIO) from one command — each skips cleanly when its prerequisite is absent. docs/cutover-gate.md is the full go/no-go checklist (reconcile clean + shadow clean incl. zero drops + gate suites + live observability). Tests: count mismatch and missing-name are flagged; identical summaries are clean. (A GitHub Actions workflow to run this stack — CGO/duckdb + postgres + minio services — is the CI-infra piece; the Makefile target is its entrypoint.)
…cy (CHD-38/37) DuckLake snapshot expiry bounds history age, not data size, so lake.signals / lake.events grow unbounded. Add an optional row-level TTL: LAKE_DECODED_RETENTION (Go duration) drives PruneDecoded, which deletes decoded rows past the window hourly from the materializer loop. Default off — deleting customer history is a product decision. The latest rollup is never pruned (current state); din's maintenance reclaims the files. docs/cutover-gate.md now documents the retention levers, the ~2× dual-run storage budget (don't shorten the bake to save storage), and notes that per-signal privilege is enforced by the GraphQL auth directives. The NATS repartitioning runbook + skew metric and MsgID sub-second precision (CHD-37) are din-side and tracked on the din branch. Test: a 30-day-old decoded row is pruned with a 7-day window; the recent row stays. Build + lint clean.
SR-3: batch GetDailyActivity. It fired one GetAggregatedSignals + one GetEventCounts per calendar day — up to 64 serialized round-trips for a 32-day window, blowing the request timeout. Replace with a single GetAggregatedSignalsForRanges + GetEventCountsForRanges pair (run concurrently), scattered by day index — the same batched path GetSegments already uses. SR-4: batch ListCloudEventsFromIndexes by subject (one raw_events query per subject, not per index) and cap the gRPC ListCloudEventsFromIndex key count at 1000 so one call can't fan out into an unbounded number of fetches. SR-5: serve GetLatestSignals from the lake.signals_latest rollup for the no-source-filter, non-location case (O(distinct-names)), matching the existing GetAllLatestSignals routing. Location names and source filters still take the full deduped scan because the rollup lacks a separate nonzero-location timestamp. SR-11: dedup raw-event fetches in SQL on the cloudevent key (subject, second-precision time, type, source, id) instead of fetching limit*2 and deduping in a Go map — the old form returned a short page when over half the window was duplicates. SR-12: default MAX_REQUEST_DURATION to 5m when unset instead of failing app startup (and the readiness probe), which crash-looped the pod. SR-16: guard GetLatestIndexAdvanced against an empty-without-error result so a lake backend can't trigger an index-out-of-range panic. SR-10: document why GetCloudEventTypeSummariesAdvanced must stay a full scan — all-time first_seen/last_seen, matching ClickHouse's unbounded summary.
SR-6: prune the per-batch signals_latest rollup refresh to the batch's hash
buckets. The DELETE and the base recompute now carry "subject_bucket IN (...)",
and both signals_latest and lake.signals are PARTITIONED BY subject_bucket, so
the recompute reads only the touched buckets' files instead of the whole table.
SR-7: count latest/summary/available reads by serving path
(dq_lake_latest_served_total{path=rollup|scan}) so the rollup-coverage gap —
source-filtered and location queries falling back to the full scan — is visible
instead of silent.
SR-14: add failure metrics + alerts that were missing — prune errors,
compaction errors, fetch-shadow mismatch/error, and a rollup-refresh-duration
gauge with a slow-refresh alert (early warning of the SR-1 O(history) cost). The
din maintenance PrometheusRule landed separately.
SR-17: prove partition pruning, not just that the column is stamped — assert the
EXPLAIN plan pushes the subject_bucket predicate to the DuckLake scan.
SR-1 (full incremental rollup) is intentionally deferred: the base insert dedups
on cloud_event_id while reads dedup on (subject,name,timestamp), so an
incremental count must replicate the read-dedup exactly or it corrupts a
customer-facing aggregate. The full recompute is correct for the default
(no-prune) config; SR-6 bounds its cost and the new gauge makes it observable.
SR-9: stop building a second query backend for the gRPC server. main() runs both app.New and CreateGRPCServer in one process, each of which opened its own duck.Service (DuckDB pool + Postgres catalog attach) and S3 client. App now exposes the eventService + buckets it already built, and CreateGRPCServer reuses them — halving per-process DuckDB/Postgres/S3 footprint. The App owns the backend lifecycle, so the gRPC path no longer returns a cleanup. SR-13: plumb a logger into NewShadowEventService — fetch-shadow mismatches were logged to a Nop logger and silently discarded, leaving only counters. Raise the shadow concurrency default 4 -> 16 (both fetch and signal shadows): at 4 the validator drops most comparisons under load, so a clean mismatch counter meant "didn't look", undermining the cutover gate. SR-18: collapse the two hand-synced HashBucket implementations — the materializer's write-side bucketing now delegates to duck.HashBucket, so write and read can never drift on the algorithm or the 256 bucket count. SR-21: PanicRecoveryMiddleware logs through the structured request logger instead of fmt.Fprintf to os.Stderr, so panics appear in the JSON log stream. SR-22: guard the materializer object store's GetObject with a generous (2 GiB) ContentLength ceiling so a corrupt/runaway object fails fast instead of being read whole into memory. SR-15 is covered without new dq code: the materializer single-writer render guard already exists (CHD-9), and the din side gained the ConsumerStaleness < SnapshotKeep boot check.
#3 events: dedup the decoded-event read path. lakeEventsDeduped collapses at-rest duplicates (QUALIFY ROW_NUMBER over subject,timestamp,name,source — ClickHouse's event ReplacingMergeTree key) so two cloud_event_ids that decode to the same logical event count once, not twice. The materializer's INSERT anti-join keeps them (its key includes cloud_event_id) and the signal read path already deduped; the event read path did not, over-counting events vs CH. #4 fetch: drop the 400-day lookback floor for subject-scoped fetches. It is a DoS guard for subject-less, id-less scans only — ClickHouse imposes no floor when given no `after`, and a subject prunes to one vehicle's files via the (subject, time) sort, so latestCloudEvent / cloudEvents now reach arbitrarily old events. A dormant vehicle whose newest event predates the window no longer wrongly looks empty. The guard stays for genuinely unbounded searches. #8 materializer: hard-error when MATERIALIZER_SHARD_COUNT>1 on the DuckLake path. Sharding is not honored there (the global ingest_progress cursor allows exactly one logical processor); refuse the config rather than silently run every replica over the full stream and roll back the CAS losers. #10 aggregations: implement inPolygon / inCircle location filters in pure SQL (haversine great-circle distance; an even-odd ray cast unrolled over the request's vertices) with no spatial-extension dependency, replacing the "not supported on duckdb backend" rejection. Mirrors ClickHouse geoDistance / pointInPolygon.
zer0stars
added a commit
that referenced
this pull request
Jul 8, 2026
…m cap (load review) (#15) * perf(dq): batch segment/daily location gap-fills into one as-of join (#8) Segment and daily-activity enrichment gap-filled each boundary's location with a separate LocationAt reverse-scan point query — O(2*segments) serial queries per request (up to thousands on a sparse-GPS vehicle with the idling candidate cap), each holding a pooled connection. Add duck.Queries.LocationsAt: one ASOF LEFT JOIN resolves the nearest non-origin fix at or before every requested boundary timestamp, index-aligned. It is exactly equivalent to per-point LocationAt — the right side excludes (0,0) fixes before deduping (subject,name,timestamp) to the lowest cloud_event_id, so the as-of match is tie-free and deterministic — proven by TestLakeQueries_LocationsAt_MatchesLocationAt across the (0,0)-skip, tie-break, and lookback-floor cases. Also add a 90d lookback floor to both LocationAt and LocationsAt so a fix-less vehicle no longer full-reverse-scans its retained partition; the floor is shared so the two paths stay equivalent. Wire GetSegments and GetDailyActivity to collect every fix-less boundary and resolve them in a single batched pass (BatchLocationAtSource), falling back to the (0,0) sentinel unchanged. TestGetSegments_BatchesLocationGapFill and TestGetDailyActivity_BatchesLocationGapFill prove O(1) location queries (one batched call, zero point calls) and correct scatter-by-index. * perf(dq): cap the idle query DuckDB on a materializer pod (#7) A materializer release always builds the query DuckDB backend (main.go serves the query HTTP/gRPC unconditionally) even though it serves no reads: the overlay disables ingress and the query fleet is a separate release with its own Service selector, so no query/fetch traffic is routed to it. That idle instance still honored DUCKDB_MEMORY_LIMIT=6GiB, so decode(6) + query(6) could reach 12GiB on an 8Gi pod and OOM. Fully skipping the query backend is entangled (main.go serves the query HTTP/gRPC and probes duckSvc for readiness unconditionally) and higher-risk, so take the lower-risk config route: add DUCKDB_QUERY_MEMORY_LIMIT, applied via queryDuckConfig ONLY when MATERIALIZER_ENABLED, to cap the idle query instance; the decode instance keeps the full DUCKDB_MEMORY_LIMIT. Set it to 1GiB in values-materializer.yaml so the two limits sum to 7GiB < 8Gi. A query pod (MATERIALIZER_ENABLED=false) ignores the override entirely. TestQueryDuckConfig_MaterializerMemoryCap pins all three cases. * feat(dq): events_latest rollup for GetEventSummaries; #5b floor deferred (#5) (a) GetEventSummaries full-history-scanned lake.events per dataSummary, with no rollup — asymmetric with the now-cheap signal side. Add lake.events_latest, a per-(subject,name) count + first/last-seen rollup maintained by the materializer exactly like signals_latest: a dirtyEventSubjects set marked post-commit, FlushEventRollup recomputing only dirty subjects bucket-chunked off the decode commit, RecomputeEventRollup for the disaster-recovery / first-create backfill, and an orphan-prune in PruneDecoded. eventRollupSelectSQL mirrors GetEventSummaries' (subject,timestamp,name,source) dedup + GROUP BY name, so the rollup is a materialized view by construction. GetEventSummaries serves from it, falling back to the base scan until the table exists (self-healing, guards a rollout where a query pod predates the materializer creating the new table). ensureSchema escalates the FIRST FlushEventRollup to a full rebuild when events_latest is created over a pre-existing lake.events (the migration case) so dormant vehicles are backfilled and never read an empty summary; a fresh catalog skips it (no snapshot churn). LAKE_REBUILD_ROLLUP_ON_BOOT rebuilds both rollups. Proven by tests/ducklake_event_rollup_test.go: end-to-end incremental via the real decode path (with a cross-cloud_event_id duplicate to prove read-dedup), full-rebuild == per-batch parity, and first-create dormant-subject backfill. (b) The signals rollup recompute timestamp floor is DEFERRED as TODO(load-review #5b): a naive floor undercounts count/first_seen (full-history aggregates) and an incremental count fold can't stay exact because the write anti-join keys on cloud_event_id, so a different-id duplicate is stored and only the read QUALIFY dedup collapses it — an incremental += would double-count it, breaking the rollup-exactness invariant the tests assert. The code carries a precise split-recency/cumulative-column design for a proven follow-up. The commit/cursor-advance exactly-once path is untouched; both rollups are materialized views maintained off the commit. * feat(dq): re-decode backfill tool + cursor-reset alert; #1c deferred (#1) (a) When the decode cursor lags past LAKE_SNAPSHOT_RETENTION, maybeRecoverExpired skips the unretained prefix WITHOUT decoding it and only cursorResetsTotal records the loss — no tool existed to recover the gap. Add BackfillTimeRange: it reads raw_events in [from, to) DIRECTLY from the base table (not the expired change feed; the rows survive din's separate row retention), re-decodes via the existing decode path, and idempotent-inserts into lake.signals/events WITHOUT touching the ingest_progress cursor (out-of-band repair). Idempotency uses the same cloud_event_id anti-join but with the UNCLAMPED [min,max] timestamp window (minMaxTime, factored out of timeRange) so re-decoding arbitrarily old data still finds and skips existing rows — the steady-state 30d probe-floor clamp would miss old duplicates and double-insert. Exposed as `dq -backfill-from <RFC3339> -backfill-to <RFC3339>` (runs once and exits), which flushes both rollups after. Proven idempotent + cursor-untouched + skipped-range-recovery by tests/ducklake_backfill_test.go. (b) The DQMaterializerCursorReset alert already existed; corrected its stale "reset to head" wording (the code skips only the unretained prefix) and made it actionable — it now names the backfill invocation to recover the gap. (c) The per-pass byte budget is DEFERRED as TODO(load-review #1c): the real OOM is one oversized snapshot (span can't drop below 1), which only intra-snapshot row-key-window pagination can bound — and that decouples the atomic insert+cursor-advance the chaos-proven exactly-once protocol relies on, so it must be re-proven under SIGKILL, not just unit-tested. The code carries the precise pagination design (page by (subject,timestamp,cloud_event_id), idempotent intermediate windows, cursor coupled only to the final window's insert). The commit/cursor-advance exactly-once path is untouched; readDelta only gained a shared scan helper (readRawByTime reuses it).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
The dq counterpart to din#2. din writes the raw layer into a DuckLake
raw_eventstable; this PR makes dq decode it and serve queries from the same lakehouse — attaching the same catalog (PostgreSQL in prod, a local file for dev/tests) that din's writer/maintainer manage.Default-safe — this merges dark.
QUERY_BACKENDdefaults toclickhouseand the materializer is off (MATERIALIZER_ENABLED=false) unless explicitly enabled, so merging changes zero runtime behavior. The DuckLake path is opt-in per environment, and the existing bucket/ClickHouse paths stay intact as the fallback until the lake proves out in shadow/prod.Query backends (
QUERY_BACKEND)internal/app/backend.goselects therepositories.Backendimplementation:clickhouse(default) — unchanged, all queries from ClickHouse.duckdb— DuckDB over hive-partitioned parquet in the bucket (the original parse-on-read approach).shadow— serve from ClickHouse, mirror each query to DuckDB in the background, compare and log divergence (the parity harness for cutover).ducklake— serve from the DuckLake catalog tables (lake.signals/lake.events).DuckLake materializer (
internal/materializer/ducklake.go)Reads new rows from din's raw layer via
ducklake_table_changes('lake','main','raw_events', cursor+1, head)filtered tochange_type='insert', decodes via model-garage, writeslake.signals/lake.events, and advances a single snapshot-id cursor inlake.ingest_progress— all in one catalog transaction. Exactly-once: concurrent materializers conflict at commit, the loser retries. No S3 LIST, no watermark file, no settle window. If the cursor falls below the catalog's oldest unexpired snapshot it resets to head, incrementsdq_materializer_cursor_resets_total, and logs loudly (the backstop if dq dies past din's staleness window).DuckLake query backend (
internal/service/duck/lake_latest.go)duck.NewLakeQueriesserves all 10repositories.Backendmethods fromlake.signals/lake.events.latest/summaryrecompute from the base table (arg_max-by-ts, on-the-fly(0,0)location filter).Consumer progress floor (the dq half of din's guarded retention)
After each materialized batch, dq attaches the catalog's side-DB as schema
metaand upserts('dq', snapshot_id, now())intometa.din_consumer_progress. din's maintainer reads that floor and never expires a snapshot at or past the slowest live consumer's cursor — so dq can't have raw rows expired out from under it while it's still catching up. This is the documented SQL contract din#2 added; this PR is the producer side.Deferred (not in this PR — sequenced after parity)
duckdb/shadowquery code. Kept as the parity reference and fallback until the lake proves out in shadow/prod — deleting now removes the safety net.QUERY_BACKEND=ducklake.voids_idis selected fromraw_eventsbut unused on the decode path (voiding is read-side;TODOinduck/raw.go).Testing
tests/ducklake_e2e_test.go) + query parity (tests/ducklake_query_test.go) — always run, no infra.TestDuckLakePostgres(tests/ducklake_pg_test.go, gated onPG_CATALOG_DSN) — proves exactly-once across two concurrent materializers against a real Postgres catalog, plus the real consumer floor.go build ./...,go test ./internal/... ./tests/,golangci-lint run ./...— all green.Config
DUCKLAKE_CATALOG_DSN(raw connection string — nopostgres:prefix; code buildsducklake:postgres:<dsn>internally),DUCKLAKE_DATA_PATH(must match din'sLAKE_DATA_PATH),MATERIALIZER_*,QUERY_BACKEND, and theDUCKDB_*engine knobs.