perf(dq): batch location gap-fills, events rollup, re-decode tool, mem 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).

Author: zer0stars

charts/dq/templates/prometheusrule.yaml

@@ -104,11 +104,14 @@ spec:
           annotations:
             summary: DuckLake materializer skipped an un-decoded snapshot range
             description: >-
-              The decode cursor lagged past LAKE_SNAPSHOT_RETENTION and was reset
-              to head, permanently skipping un-decoded raw events
-              ({{ "{{ $value }}" }} resets; span in
-              dq_materializer_last_cursor_reset_gap_snapshots). Backfill the
-              skipped range, then raise retention or decode throughput.
+              The decode cursor lagged past LAKE_SNAPSHOT_RETENTION; the unretained
+              prefix was skipped WITHOUT decoding, so those raw events never reached
+              lake.signals/events ({{ "{{ $value }}" }} resets; skipped span in
+              dq_materializer_last_cursor_reset_gap_snapshots). RECOVER the gap with
+              the backfill tool — run the dq image with
+              `-backfill-from <RFC3339> -backfill-to <RFC3339>` covering the skipped
+              window (idempotent; safe while the materializer is up) — then raise
+              LAKE_SNAPSHOT_RETENTION or decode throughput so it can't recur.
         # Decode-position spread as a single number (CHD-14). NOT an exact
         # backlog: head is the catalog-global max snapshot id, which includes
         # this decoder's own signals/events/rollup snapshots and din's

charts/dq/values-materializer.yaml

@@ -49,9 +49,18 @@ env:
   # duration; ~1y). Empty lets them grow unbounded.
   LAKE_DECODED_RETENTION: 8760h
   # Decode + rollup recompute are memory-heavy (one pod). Size memory_limit to
-  # ~75% of the pod memory limit; threads to the CPU limit.
+  # ~75% of the pod memory limit; threads to the CPU limit. This is the DECODE
+  # instance's budget.
   DUCKDB_MEMORY_LIMIT: 6GiB
   DUCKDB_THREADS: 4
+  # This release always ALSO builds a query DuckDB instance (main.go serves the
+  # query HTTP/gRPC unconditionally), but it receives no traffic here: no ingress,
+  # and the query fleet is a separate release with its own Service. Without this it
+  # would honor the 6GiB DUCKDB_MEMORY_LIMIT too, so decode(6) + query(6) = 12GiB
+  # can OOM the 8Gi pod (finding #7). Cap the idle query instance at 1GiB so the two
+  # limits sum to 7GiB < 8Gi; only a query pod (MATERIALIZER_ENABLED=false) ignores
+  # this and keeps the full budget.
+  DUCKDB_QUERY_MEMORY_LIMIT: 1GiB
 
 resources:
   limits:

cmd/dq/main.go

@@ -45,12 +45,23 @@ func main() {
 	runnerGroup, runnerCtx := errgroup.WithContext(mainCtx)
 
 	settingsFile := flag.String("settings", "settings.yaml", "settings file")
+	// Backfill mode (finding #1a): a one-shot re-decode of a raw_events time range into
+	// the decoded lake tables, for a range the decode loop permanently skipped on cursor
+	// expiry (DQMaterializerCursorReset). Both flags RFC3339; setting either runs the
+	// backfill and exits instead of starting the servers/decode loop. Idempotent.
+	backfillFrom := flag.String("backfill-from", "", "RFC3339 start (inclusive) of a raw_events range to re-decode, then exit")
+	backfillTo := flag.String("backfill-to", "", "RFC3339 end (exclusive) of a raw_events range to re-decode, then exit")
 	flag.Parse()
 
 	cfg, err := settings.LoadConfig[config.Settings](*settingsFile)
 	if err != nil {
 		logger.Fatal().Err(err).Msg("Couldn't load settings.")
 	}
+
+	if *backfillFrom != "" || *backfillTo != "" {
+		runBackfillAndExit(cfg, *backfillFrom, *backfillTo, logger)
+		return
+	}
 	// The shared env loader silently leaves a field zero on a malformed value (it swallows
 	// per-field parse errors), so boot-critical numerics never fail LoadConfig — validate
 	// them loud here (a zero port black-holes traffic; a zero chain id mis-decodes DIDs).
@@ -103,6 +114,25 @@ func main() {
 	logger.Info().Msg("Server shut down.")
 }
 
+// runBackfillAndExit parses the RFC3339 backfill window, runs the one-shot re-decode
+// (finding #1a), and exits. Both bounds are required and from must precede to.
+func runBackfillAndExit(cfg config.Settings, fromStr, toStr string, logger zerolog.Logger) {
+	if fromStr == "" || toStr == "" {
+		logger.Fatal().Msg("backfill requires both -backfill-from and -backfill-to (RFC3339)")
+	}
+	from, err := time.Parse(time.RFC3339, fromStr)
+	if err != nil {
+		logger.Fatal().Err(err).Str("backfill-from", fromStr).Msg("invalid -backfill-from (want RFC3339)")
+	}
+	to, err := time.Parse(time.RFC3339, toStr)
+	if err != nil {
+		logger.Fatal().Err(err).Str("backfill-to", toStr).Msg("invalid -backfill-to (want RFC3339)")
+	}
+	if err := app.RunBackfill(cfg, from, to, logger); err != nil {
+		logger.Fatal().Err(err).Msg("backfill failed")
+	}
+}
+
 // maxHTTPBodyBytes caps a request body on the public query surface. GraphQL
 // queries/variables are small; this blunts a single oversized POST.
 const maxHTTPBodyBytes = 4 << 20 // 4 MiB

internal/app/backend.go

@@ -61,6 +61,23 @@ func duckConfigFromSettings(settings *config.Settings) duck.Config {
 	}
 }
 
+// queryDuckConfig is the DuckDB config for the QUERY backend. It is
+// duckConfigFromSettings with one materializer-pod adjustment: on a materializer
+// release (MATERIALIZER_ENABLED) the query backend is always built but serves no
+// traffic (no ingress; the query fleet is a separate release), yet still honors
+// DUCKDB_MEMORY_LIMIT — so two same-limit DuckDB instances can co-reside on one pod
+// and OOM (finding #7). When DUCKDB_QUERY_MEMORY_LIMIT is set on such a pod, cap this
+// idle query instance with it so it plus the decode instance sum under the pod limit;
+// the decode instance (startDuckLakeMaterializer) keeps the full DUCKDB_MEMORY_LIMIT.
+// On a query pod the override is inert, so the query fleet is unchanged.
+func queryDuckConfig(settings *config.Settings) duck.Config {
+	cfg := duckConfigFromSettings(settings)
+	if settings.MaterializerEnabled && settings.DuckDBQueryMemoryLimit != "" {
+		cfg.DuckDBMemoryLimit = settings.DuckDBQueryMemoryLimit
+	}
+	return cfg
+}
+
 // isLocalBucket reports whether the parquet bucket points at the local
 // filesystem (file:// URL or absolute path) instead of S3, mirroring how
 // duck.Service interprets its Bucket setting.
@@ -79,7 +96,7 @@ func newQueryBackend(settings *config.Settings, logger zerolog.Logger) (reposito
 	if settings.DuckLakeCatalogDSN == "" {
 		return nil, nil, nil, fmt.Errorf("DUCKLAKE_CATALOG_DSN is empty: the DuckLake catalog is required for the query backend")
 	}
-	duckSvc, err := duck.NewService(duckConfigFromSettings(settings))
+	duckSvc, err := duck.NewService(queryDuckConfig(settings))
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("couldn't create DuckDB service: %w", err)
 	}
@@ -144,13 +161,38 @@ func startMaterializer(settings *config.Settings, logger zerolog.Logger) (func()
 // owns its own DuckDB service (catalog attached) for the lifetime of the
 // loop; the query backend opens a separate one.
 func startDuckLakeMaterializer(settings *config.Settings, pollInterval time.Duration, logger zerolog.Logger) (func(), error) {
+	duckSvc, mat, runner, err := buildDuckLakeMaterializer(settings, pollInterval, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	// rebuildRollup is the opt-in disaster-recovery rebuild (LAKE_REBUILD_ROLLUP_ON_BOOT):
+	// the per-batch recompute only touches a batch's subjects, so a dropped/truncated
+	// rollup leaves dormant vehicles missing until rebuilt from the full base. It runs
+	// in the loop goroutine BEFORE processing (never concurrently with it) — not
+	// synchronously at boot — so this O(history) scan can't outlast the liveness probe
+	// and CrashLoop the pod; a failure is logged and the loop proceeds (the per-batch
+	// recompute still heals active vehicles).
+	stop := runMaterializerLoop(runner, mat, settings.LakeRebuildRollupOnBoot, logger)
+	return func() {
+		stop()
+		_ = duckSvc.Close()
+	}, nil
+}
+
+// buildDuckLakeMaterializer constructs the materializer's DuckDB service, the
+// DuckLakeMaterializer (blob store + cipher + temp dir), and the Runner (the decode
+// surface) — the shared setup behind both the live decode loop (startDuckLakeMaterializer)
+// and the one-shot re-decode tool (RunBackfill, finding #1a). On error it closes the
+// service so the caller has nothing to clean up; on success the caller owns duckSvc.
+func buildDuckLakeMaterializer(settings *config.Settings, pollInterval time.Duration, logger zerolog.Logger) (*duck.Service, *materializer.DuckLakeMaterializer, *materializer.Runner, error) {
 	// Sharding is not honored on the DuckLake path — readDelta reads the whole
 	// raw_events delta and the single global ingest_progress cursor makes exactly
 	// one logical processor (extra replicas just lose the cursor CAS and roll
 	// back). Refuse the config rather than silently ignore it: run the
 	// materializer as a single replicaCount=1 release (SR review #8).
 	if settings.MaterializerShardCount > 1 {
-		return nil, fmt.Errorf(
+		return nil, nil, nil, fmt.Errorf(
 			"MATERIALIZER_SHARD_COUNT=%d is not supported on the DuckLake path: the global ingest_progress cursor allows only one materializer; run a single replicaCount=1 release",
 			settings.MaterializerShardCount)
 	}
@@ -162,20 +204,20 @@ func startDuckLakeMaterializer(settings *config.Settings, pollInterval time.Dura
 	cfg.MetricsPoolLabel = "materializer" // separate dq_db_pool_* series from the query pool (H4)
 	duckSvc, err := duck.NewService(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("creating DuckLake service: %w", err)
+		return nil, nil, nil, fmt.Errorf("creating DuckLake service: %w", err)
 	}
 	mat, err := materializer.NewDuckLakeMaterializer(context.Background(), duckSvc.DB(), logger)
 	if err != nil {
 		_ = duckSvc.Close()
-		return nil, fmt.Errorf("creating DuckLake materializer: %w", err)
+		return nil, nil, nil, fmt.Errorf("creating DuckLake materializer: %w", err)
 	}
 	// Resolve externalized blob payloads from the same bucket the fetch path
 	// presigns/downloads (settings.BlobBucket): din writes payloads larger
 	// than the inline threshold to a blob and leaves only the key on the row.
 	blobCipher, err := blobcrypt.NewCipher(settings.BlobEncryptionKey)
 	if err != nil {
 		_ = duckSvc.Close()
-		return nil, fmt.Errorf("blob cipher: %w", err)
+		return nil, nil, nil, fmt.Errorf("blob cipher: %w", err)
 	}
 	mat = mat.WithBlobStore(s3ClientFromSettings(settings), settings.BlobBucket).
 		WithBlobCipher(blobCipher).
@@ -186,15 +228,15 @@ func startDuckLakeMaterializer(settings *config.Settings, pollInterval time.Dura
 		decodedRetention, err = time.ParseDuration(settings.LakeDecodedRetention)
 		if err != nil {
 			_ = duckSvc.Close()
-			return nil, fmt.Errorf("invalid LAKE_DECODED_RETENTION %q: %w", settings.LakeDecodedRetention, err)
+			return nil, nil, nil, fmt.Errorf("invalid LAKE_DECODED_RETENTION %q: %w", settings.LakeDecodedRetention, err)
 		}
 	}
 	var rollupInterval time.Duration
 	if settings.MaterializerRollupInterval != "" {
 		rollupInterval, err = time.ParseDuration(settings.MaterializerRollupInterval)
 		if err != nil {
 			_ = duckSvc.Close()
-			return nil, fmt.Errorf("invalid MATERIALIZER_ROLLUP_INTERVAL %q: %w", settings.MaterializerRollupInterval, err)
+			return nil, nil, nil, fmt.Errorf("invalid MATERIALIZER_ROLLUP_INTERVAL %q: %w", settings.MaterializerRollupInterval, err)
 		}
 	}
 	runner := materializer.New(materializer.Config{
@@ -206,19 +248,46 @@ func startDuckLakeMaterializer(settings *config.Settings, pollInterval time.Dura
 		DecodedRetention:  decodedRetention,
 		BackfillMode:      settings.MaterializerBackfillMode,
 	}, logger).WithDuckLake(mat)
+	return duckSvc, mat, runner, nil
+}
 
-	// rebuildRollup is the opt-in disaster-recovery rebuild (LAKE_REBUILD_ROLLUP_ON_BOOT):
-	// the per-batch recompute only touches a batch's subjects, so a dropped/truncated
-	// rollup leaves dormant vehicles missing until rebuilt from the full base. It runs
-	// in the loop goroutine BEFORE processing (never concurrently with it) — not
-	// synchronously at boot — so this O(history) scan can't outlast the liveness probe
-	// and CrashLoop the pod; a failure is logged and the loop proceeds (the per-batch
-	// recompute still heals active vehicles).
-	stop := runMaterializerLoop(runner, mat, settings.LakeRebuildRollupOnBoot, logger)
-	return func() {
-		stop()
-		_ = duckSvc.Close()
-	}, nil
+// RunBackfill re-decodes raw_events in [from, to) into the decoded lake tables and
+// exits — the operator-run repair for a range the decode loop permanently skipped on
+// cursor expiry (finding #1a; alerted by DQMaterializerCursorReset). It is idempotent
+// (the same cloud_event_id anti-join), so it is safe to re-run and safe to run while
+// the live materializer is up. It registers the vendor decode modules, decodes the
+// range, then flushes both rollups so latest/summary reflect the backfilled rows.
+func RunBackfill(settings config.Settings, from, to time.Time, logger zerolog.Logger) error {
+	if settings.DuckLakeCatalogDSN == "" {
+		return fmt.Errorf("DUCKLAKE_CATALOG_DSN is empty: the DuckLake catalog is required for backfill")
+	}
+	materializer.RegisterVendorModules(materializer.VendorConfig{
+		ChainID:               settings.DIMORegistryChainID,
+		VehicleNFTAddress:     common.HexToAddress(settings.VehicleNFTAddress),
+		AftermarketNFTAddress: common.HexToAddress(settings.AftermarketNFTAddress),
+		SyntheticNFTAddress:   common.HexToAddress(settings.SyntheticNFTAddress),
+	})
+	duckSvc, mat, runner, err := buildDuckLakeMaterializer(&settings, 0, logger)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = duckSvc.Close() }()
+
+	ctx := context.Background()
+	logger.Info().Time("from", from).Time("to", to).Msg("backfill: re-decoding raw_events range")
+	n, err := mat.BackfillTimeRange(ctx, runner, from, to)
+	if err != nil {
+		return fmt.Errorf("backfill decode: %w", err)
+	}
+	// Refresh the rollups for the subjects the backfill touched (dirtied above).
+	if ferr := runner.FlushRollup(ctx); ferr != nil {
+		logger.Error().Err(ferr).Msg("backfill: signals_latest flush failed; rerun with LAKE_REBUILD_ROLLUP_ON_BOOT if latest/summary looks stale")
+	}
+	if ferr := runner.FlushEventRollup(ctx); ferr != nil {
+		logger.Error().Err(ferr).Msg("backfill: events_latest flush failed; rerun with LAKE_REBUILD_ROLLUP_ON_BOOT if event summaries look stale")
+	}
+	logger.Info().Int("raw_events", n).Msg("backfill complete")
+	return nil
 }
 
 // runMaterializerLoop runs runner.Run in a goroutine and returns a stop
@@ -229,14 +298,21 @@ func runMaterializerLoop(runner *materializer.Runner, mat *materializer.DuckLake
 	go func() {
 		defer close(done)
 		if rebuildRollup {
-			logger.Info().Msg("LAKE_REBUILD_ROLLUP_ON_BOOT set: rebuilding signals_latest from full base (may take a while on deep history)")
+			logger.Info().Msg("LAKE_REBUILD_ROLLUP_ON_BOOT set: rebuilding signals_latest + events_latest from full base (may take a while on deep history)")
 			if err := mat.RecomputeRollup(ctx); err != nil {
 				// Non-fatal: log and proceed — crashing here would CrashLoop the
 				// pod; the per-batch recompute still heals active vehicles.
 				logger.Error().Err(err).Msg("signals_latest rebuild failed; continuing with the normal loop")
 			} else {
 				logger.Info().Msg("signals_latest rebuild complete")
 			}
+			// Rebuild the events rollup too (finding #5a); independent of the signals
+			// rebuild so one failing doesn't skip the other.
+			if err := mat.RecomputeEventRollup(ctx); err != nil {
+				logger.Error().Err(err).Msg("events_latest rebuild failed; continuing with the normal loop")
+			} else {
+				logger.Info().Msg("events_latest rebuild complete")
+			}
 		}
 		if err := runner.Run(ctx); err != nil {
 			// Run returns an error only when the decode loop is durably broken (the

internal/app/backend_test.go

@@ -0,0 +1,48 @@
+package app
+
+import (
+	"testing"
+
+	"github.com/DIMO-Network/dq/internal/config"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestQueryDuckConfig_MaterializerMemoryCap proves finding #7's fix: on a
+// materializer pod the always-built (but idle) query DuckDB instance is capped by
+// DUCKDB_QUERY_MEMORY_LIMIT so it plus the decode instance sum under the pod limit,
+// while a query pod is unaffected.
+func TestQueryDuckConfig_MaterializerMemoryCap(t *testing.T) {
+	t.Parallel()
+	base := config.Settings{
+		DuckLakeCatalogDSN: "postgres://x", // non-empty so this models a real config
+		DuckDBMemoryLimit:  "6GiB",
+	}
+
+	t.Run("query pod ignores the override", func(t *testing.T) {
+		s := base
+		s.MaterializerEnabled = false
+		s.DuckDBQueryMemoryLimit = "1GiB"
+		assert.Equal(t, "6GiB", queryDuckConfig(&s).DuckDBMemoryLimit,
+			"a query pod keeps the full DUCKDB_MEMORY_LIMIT")
+	})
+
+	t.Run("materializer pod caps the idle query instance", func(t *testing.T) {
+		s := base
+		s.MaterializerEnabled = true
+		s.DuckDBQueryMemoryLimit = "1GiB"
+		assert.Equal(t, "1GiB", queryDuckConfig(&s).DuckDBMemoryLimit,
+			"the idle query instance on a materializer pod must use the lower cap")
+		// The decode instance keeps the full budget (it uses duckConfigFromSettings
+		// directly, not queryDuckConfig).
+		assert.Equal(t, "6GiB", duckConfigFromSettings(&s).DuckDBMemoryLimit,
+			"the decode instance keeps DUCKDB_MEMORY_LIMIT")
+	})
+
+	t.Run("materializer pod without an override keeps the full budget", func(t *testing.T) {
+		s := base
+		s.MaterializerEnabled = true
+		s.DuckDBQueryMemoryLimit = ""
+		assert.Equal(t, "6GiB", queryDuckConfig(&s).DuckDBMemoryLimit,
+			"no override set → unchanged (opt-in)")
+	})
+}

internal/config/settings.go

@@ -78,8 +78,17 @@ type Settings struct {
 	// read-only reader role. Empty reads the primary DuckLakeCatalogDSN.
 	DuckLakeCatalogReadDSN string `yaml:"DUCKLAKE_CATALOG_READ_DSN"`
 	// DuckDB parse-on-read query engine (maps into duck.Config).
-	DuckDBMemoryLimit   string `yaml:"DUCKDB_MEMORY_LIMIT"`
-	DuckDBThreads       int    `yaml:"DUCKDB_THREADS"`
+	DuckDBMemoryLimit string `yaml:"DUCKDB_MEMORY_LIMIT"`
+	// DuckDBQueryMemoryLimit, when set, overrides DuckDBMemoryLimit for the QUERY
+	// DuckDB instance ONLY ON A MATERIALIZER POD (MATERIALIZER_ENABLED). The
+	// materializer release serves no query traffic (no ingress; the query fleet is a
+	// separate release), so its always-constructed query backend is idle — but it
+	// still honors DUCKDB_MEMORY_LIMIT, so two 6GiB DuckDB instances can co-reside on
+	// an 8Gi pod and OOM (finding #7). Lowering just the idle query instance makes the
+	// two limits sum under the pod limit while the decode instance keeps its full
+	// budget. Ignored on a query pod (MATERIALIZER_ENABLED=false).
+	DuckDBQueryMemoryLimit string `yaml:"DUCKDB_QUERY_MEMORY_LIMIT"`
+	DuckDBThreads          int    `yaml:"DUCKDB_THREADS"`
 	DuckDBExtensionDir  string `yaml:"DUCKDB_EXTENSION_DIR"`
 	DuckDBTempDirectory string `yaml:"DUCKDB_TEMP_DIRECTORY"`
 	DuckDBMaxConns      int    `yaml:"DUCKDB_MAX_CONNS"`