feat(data-warehouse): duckgres sink backfill primer

[EDsCODE]

Problem

When the duckgres batch sink (#59251) is enabled for a team, pre-existing incremental/append schemas get a permanently partial duckgres table: their history lives in Delta, the queue only retains ~7 days of batches, and the sink's per-batch stream never replays it. The legacy fix — DuckLakeCopyDataImportsWorkflow's full-table CREATE OR REPLACE … AS delta_scan(...) — hangs/OOMs on huge tables: one giant statement in one duckgres worker session, bounded by the session memory budget, with no checkpoint or resume.

Design

Backfill becomes normal sink traffic instead of one heroic statement:

flag flip → planner pins a Delta snapshot → ~1GiB parquet chunks
         → synthetic queue run (pre-marked delta-succeeded; invisible to the Delta consumer)
         → normal consumer applies chunks into <table>__backfill (apply markers = exactly-once + resume)
         → last chunk atomically DROP+RENAMEs over the live table → schema PRIMED
         → live runs (queued behind via the cross-run gate) MERGE on top

• DuckgresSinkSchemaState (app DB): PENDING_BACKFILL → BACKFILLING → PRIMED (+NEEDS_RESYNC). Live batches for a schema are blocked until PRIMED — including schemas with no state row yet, so there is no partial-history window at flag flip. Full-refresh/new/CDC schemas go straight to PRIMED.
• Chunk planning (consumer maintenance tick, deltalake metadata only): when the snapshot has no deletion vectors the chunks are the Delta table's own live parquet files (zero data movement); otherwise a streamed, memory-bounded staging pass writes resolved chunks to duckgres_backfill_staging/ (outside the extract-retention sweep). Re-planning inserts only missing chunks, so queue-partition drops cost no progress.
• The pre-succeeded trick: synthetic sourcebatch rows get a sourcebatchstatus='succeeded' row up front — the Delta consumer never claims them, the duckgres fetch sees them immediately. Zero Delta-side changes.
• Atomic swap: last chunk's transaction = INSERT + DROP TABLE IF EXISTS live + ALTER TABLE __backfill RENAME TO live + apply marker. Rename is metadata-only in DuckLake (integration-tested in duckgres: tests/integration/ddl_test.go:144), so swap cost is size-independent and readers never see a half-built table.
• Interactions audited: a live full refresh superseding the backfill flips the schema straight to PRIMED (the replace yields the complete table anyway); backfill runs are is_resume=true so they can never act as supersede replace-heads; MAX_CONCURRENT_BACKFILLS_PER_ORG=1 keeps org workers usable.
• Ops: reset_duckgres_failed_runs --replan-backfill <schema-id>; duckgres_backfill_schemas{state} gauge; progress denormalized on the state row.

How did you test this code?

• Unit: chunk grouping boundaries, processor backfill branch (create / swap+PRIMED / idempotent no-op), planner fail-safe (consumer claims nothing until the first successful planner pass).
• DB-backed (CI): blocked-schema gate excludes live batches and passes backfill batches; backfill runs excluded from supersede's replace-head set.
• Not yet run: end-to-end against a dev duckgres (needs the stacked PRs deployed); the scale rehearsal from the design doc is a pre-rollout step.

Stacked on

#59251 (duckgres-v3-consumer) — uses its queue SQL, apply markers, enablement, and processor machinery.

Publish to changelog?

do not publish to changelog

🤖 Generated with Claude Code

[github-actions]

Migration SQL Changes

Hey 👋, we've detected some migrations on this PR. Here's the SQL output for each migration, make sure they make sense:

posthog/migrations/1245_duckgres_sink_schema_state.py

BEGIN;
--
-- Create model DuckgresSinkSchemaState
--
CREATE TABLE "posthog_duckgressinkschemastate" ("created_at" timestamp with time zone NOT NULL, "updated_at" timestamp with time zone NULL, "id" uuid NOT NULL PRIMARY KEY, "schema_id" uuid NOT NULL UNIQUE, "state" varchar(32) NOT NULL, "snapshot_version" bigint NULL, "plan_cutoff" timestamp with time zone NULL, "backfill_run_uuid" varchar(200) NULL, "chunk_count" integer NULL, "chunks_applied" integer NOT NULL, "last_error" text NULL, "created_by_id" integer NULL, "team_id" integer NOT NULL);
--
-- Create index duckgres_sink_team_state_idx on field(s) team, state of model duckgressinkschemastate
--
CREATE INDEX "duckgres_sink_team_state_idx" ON "posthog_duckgressinkschemastate" ("team_id", "state");
CREATE INDEX "posthog_duckgressinkschemastate_created_by_id_ac69def6" ON "posthog_duckgressinkschemastate" ("created_by_id");
CREATE INDEX "posthog_duckgressinkschemastate_team_id_8014d2d7" ON "posthog_duckgressinkschemastate" ("team_id");
COMMIT;

Last updated: 2026-06-26 18:52 UTC (8c8f6c0)

[github-actions]

🔍 Migration Risk Analysis

We've analyzed your migrations for potential risks.

Summary: 1 Safe | 0 Needs Review | 0 Blocked

✅ Safe

Brief or no lock, backwards compatible

posthog.1245_duckgres_sink_schema_state
  └─ #1 ✅ CreateModel
     Creating new table is safe
     model: DuckgresSinkSchemaState
  │
  └──> ℹ️  INFO:
       ℹ️  Skipped operations on newly created tables (empty tables
       don't cause lock contention).

Last updated: 2026-06-26 18:52 UTC (8c8f6c0)

[greptile-apps]

Comments Outside Diff (1)

1. posthog/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_backfill.py, line 122-147 (link)

   Prefer @pytest.mark.parametrize for the chunk-grouping cases

   The three TestChunkGrouping methods test variations of the same function with different inputs and expected outputs — this is the textbook parametrize use case. The codebase preference is parametrised tests for exactly this pattern.

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: posthog/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_backfill.py
   Line: 122-147
   
   Comment:
   **Prefer `@pytest.mark.parametrize` for the chunk-grouping cases**
   
   The three `TestChunkGrouping` methods test variations of the same function with different inputs and expected outputs — this is the textbook parametrize use case. The codebase preference is parametrised tests for exactly this pattern.
   
   How can I resolve this? If you propose a fix, please make it concise.

   Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

Prompt To Fix All With AI

Fix the following 3 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 3
posthog/temporal/data_imports/pipelines/pipeline_v3/duckgres/backfill.py:430-482
**Non-atomic chunk enqueue — orphaned batches after partial failure**

`_enqueue_chunks` runs with `autocommit=True`, so the `INSERT INTO {BATCH_TABLE}` and `INSERT INTO {DELTA_STATUS_TABLE}` are separate transactions. If the process crashes or the connection drops between the two, the batch row lands in `BATCH_TABLE` without a `'succeeded'` entry in `DELTA_STATUS_TABLE`.

On the next planner pass, the idempotency check (`if chunk.index in existing`) queries only `BATCH_TABLE`, so this chunk is considered already-enqueued and is skipped. The duckgres consumer only fetches delta-succeeded batches, so the chunk is permanently invisible. The backfill stalls at `chunks_applied < chunk_count` indefinitely, with no error surfaced.

The fix is to wrap both inserts in `with conn.transaction():` for each chunk iteration.

### Issue 2 of 3
posthog/temporal/data_imports/pipelines/pipeline_v3/duckgres/processor.py:312-315
**`mark_primed` outside the committed transaction — permanently stuck BACKFILLING**

`mark_primed` runs after `with conn.transaction()` exits, so the duckgres swap (DROP + RENAME + apply marker) can commit while the app-DB state update fails. When `mark_primed` raises:

1. The exception propagates through `process_batch`; the consumer marks this batch `'failed'`.
2. On retry, `_has_duckgres_batch_applied` → `True` (apply marker committed in duckgres), so the processor returns at the early-exit guard — `mark_primed` is never reached again.
3. `after_batch_processed` runs and marks the batch succeeded in the queue; the reconciler finds no failed batches for this `run_uuid`.
4. `_reconcile_backfilling` only transitions to `PRIMED` when a failed batch's error contains `"superseded"`; there is no path for `chunks_applied == chunk_count → PRIMED`.

The schema is permanently stuck in `BACKFILLING` with all live batches blocked, no `last_error` set, and no automated recovery. The only escape is operator intervention via `reset_duckgres_failed_runs --replan-backfill`.

Consider adding an auto-PRIME transition in `_reconcile_backfilling` when `applied[0] >= state.chunk_count` (and `state.chunk_count` is set), or catching the `mark_primed` error and scheduling a retry rather than propagating.

### Issue 3 of 3
posthog/temporal/data_imports/pipelines/pipeline_v3/duckgres/test_backfill.py:122-147
**Prefer `@pytest.mark.parametrize` for the chunk-grouping cases**

The three `TestChunkGrouping` methods test variations of the same function with different inputs and expected outputs — this is the textbook parametrize use case. The codebase preference is parametrised tests for exactly this pattern.

_{Reviews (1): Last reviewed commit: "feat(data-warehouse): duckgres sink back..." | Re-trigger Greptile}

[greptile-apps]

mark_primed outside the committed transaction — permanently stuck BACKFILLING

mark_primed runs after with conn.transaction() exits, so the duckgres swap (DROP + RENAME + apply marker) can commit while the app-DB state update fails. When mark_primed raises:

1. The exception propagates through process_batch; the consumer marks this batch 'failed'.
2. On retry, _has_duckgres_batch_applied → True (apply marker committed in duckgres), so the processor returns at the early-exit guard — mark_primed is never reached again.
3. after_batch_processed runs and marks the batch succeeded in the queue; the reconciler finds no failed batches for this run_uuid.
4. _reconcile_backfilling only transitions to PRIMED when a failed batch's error contains "superseded"; there is no path for chunks_applied == chunk_count → PRIMED.

The schema is permanently stuck in BACKFILLING with all live batches blocked, no last_error set, and no automated recovery. The only escape is operator intervention via reset_duckgres_failed_runs --replan-backfill.

Consider adding an auto-PRIME transition in _reconcile_backfilling when applied[0] >= state.chunk_count (and state.chunk_count is set), or catching the mark_primed error and scheduling a retry rather than propagating.

Prompt To Fix With AI

This is a comment left during a code review.
Path: posthog/temporal/data_imports/pipelines/pipeline_v3/duckgres/processor.py
Line: 312-315

Comment:
**`mark_primed` outside the committed transaction — permanently stuck BACKFILLING**

`mark_primed` runs after `with conn.transaction()` exits, so the duckgres swap (DROP + RENAME + apply marker) can commit while the app-DB state update fails. When `mark_primed` raises:

1. The exception propagates through `process_batch`; the consumer marks this batch `'failed'`.
2. On retry, `_has_duckgres_batch_applied` → `True` (apply marker committed in duckgres), so the processor returns at the early-exit guard — `mark_primed` is never reached again.
3. `after_batch_processed` runs and marks the batch succeeded in the queue; the reconciler finds no failed batches for this `run_uuid`.
4. `_reconcile_backfilling` only transitions to `PRIMED` when a failed batch's error contains `"superseded"`; there is no path for `chunks_applied == chunk_count → PRIMED`.

The schema is permanently stuck in `BACKFILLING` with all live batches blocked, no `last_error` set, and no automated recovery. The only escape is operator intervention via `reset_duckgres_failed_runs --replan-backfill`.

Consider adding an auto-PRIME transition in `_reconcile_backfilling` when `applied[0] >= state.chunk_count` (and `state.chunk_count` is set), or catching the `mark_primed` error and scheduling a retry rather than propagating.

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

_{Reviews (2): Last reviewed commit: "Merge branch 'master' into eric/duckgres..." | Re-trigger Greptile}

[tests-posthog]

⏭️ Skipped snapshot commit because branch advanced to 30e93de while workflow was testing 581228d.

The new commit will trigger its own snapshot update workflow.

If you expected this workflow to succeed: This can happen due to concurrent commits. To get a fresh workflow run, either:

• Merge master into your branch, or
• Push an empty commit: git commit --allow-empty -m 'trigger CI' && git push

[tests-posthog]

Query snapshots: Backend query snapshots updated

Changes: 1 snapshots (1 modified, 0 added, 0 deleted)

What this means:

• Query snapshots have been automatically updated to match current output
• These changes reflect modifications to database queries or schema

Next steps:

• Review the query changes to ensure they're intentional
• If unexpected, investigate what caused the query to change

Review snapshot changes →

[tests-posthog]

Query snapshots: Backend query snapshots updated

Changes: 1 snapshots (1 modified, 0 added, 0 deleted)

What this means:

• Query snapshots have been automatically updated to match current output
• These changes reflect modifications to database queries or schema

Next steps:

• Review the query changes to ensure they're intentional
• If unexpected, investigate what caused the query to change

Review snapshot changes →

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-06-26 19:49 UTC	Run
prod-us	✅ Deployed	2026-06-26 20:50 UTC	Run
prod-eu	✅ Deployed	2026-06-26 21:13 UTC	Run