feat(managed-warehouse): duckling schema prep for streamed persons

posthog/dags/PERSONS_STREAMING_MIGRATION.md

@@ -0,0 +1,166 @@
+# Persons streaming migration runbook
+
+How a duckling's persons data moves from the Dagster batch backfill
+(`duckling_persons_backfill`) to streaming replication (millpond + viaduck).
+Read this alongside [README_DUCKLINGS.md](README_DUCKLINGS.md), which covers
+the batch system this replaces.
+
+Status: **in progress**. This file is the coordination point for the
+cross-repo work; the per-repo pieces are linked at the bottom.
+
+## Why
+
+The batch job re-exports persons daily by `_timestamp` (Kafka ingestion time)
+and replaces the day's partition: ClickHouse `FINAL` join, S3 parquet,
+ranged `DELETE`, `ducklake_add_data_files`. It works, but data is up to a day
+stale, every re-export re-reads full person state, and the delete-then-register
+window is not atomic for readers. Streaming lands person changes in the
+duckling within minutes.
+
+## What changes shape
+
+The batch table is **denormalized**: one row per `distinct_id`, person
+properties stamped on, produced by a `person FINAL ⋈ person_distinct_id2
+FINAL` join. Neither millpond (one Kafka topic to one table, insert-only) nor
+viaduck (row-level CDC routing, no joins) can produce that shape, so a
+streamed duckling carries three objects instead of one table:
+
+| Object | Name (suffixed team `ab12`) | Written by | Shape |
+|---|---|---|---|
+| Raw persons | `persons_ab12_raw` | viaduck full_cdc, upsert key `(team_id, id)` | Latest version per person, `is_deleted=1` tombstones included |
+| Raw distinct-ids | `persons_distinct_ids_ab12_raw` | viaduck full_cdc, upsert key `(team_id, distinct_id)` | Latest mapping per distinct_id, tombstones included |
+| Denormalized view | `persons_ab12` (canonical, post-cutover) | nobody (view) | Exactly the batch table's columns, tombstones filtered |
+
+Two things to internalize before touching a team:
+
+1. **Tombstones are rows, not deletes.** The Kafka person changelogs signal
+   deletion with `is_deleted=1` upserts; the source lake millpond lands is
+   append-only, so viaduck only ever sees inserts and upserts them by key.
+   Deleted persons stay physically present in the raw tables and the view
+   filters them. Readers querying raw tables directly must filter
+   `is_deleted` themselves.
+2. **Merges converge through the mapping table.** A person merge moves
+   distinct_ids between persons: the distinct-ids changelog emits new
+   mapping versions, viaduck upserts them, and the view's join follows the
+   new mapping. No cross-tenant routing is involved because merges stay
+   within a team.
+
+## Pipeline architecture
+
+```
+clickhouse_person topic ─────────┐
+                                 ├─► millpond (2 StatefulSets, team allowlist
+clickhouse_person_distinct_id ───┘    from the duckgres control plane)
+                                             │
+                                             ▼
+                              shared changelog DuckLake (megaduck)
+                                             │
+               viaduck full_cdc, 2 pipelines, route by team_id
+               (static destinations during migration; discovery later)
+                                             │
+                                             ▼
+                per-duckling persons_<suffix>_raw + persons_distinct_ids_<suffix>_raw
+```
+
+The full historical backfill **stays batch**. Kafka retention bounds how far
+back millpond can read, and viaduck discovery initializes destinations at the
+source head by design. Streaming replaces the daily top-up sensor, not the
+initial load.
+
+## Per-team cutover
+
+Prerequisites: the millpond persons consumers are already running (they land
+the `person` / `person_distinct_id` changelogs on megaduck), and the viaduck
+persons pipelines are deployed (charts `viaduck-persons` /
+`viaduck-persons-distinct-ids` ApplicationSets).
+
+1. **Prep the duckling schema.** Run any `duckling_persons_backfill_job`
+   partition for the team with:
+
+   ```yaml
+   ops:
+     duckling_persons_backfill:
+       config:
+         create_persons_streaming_schema: true
+         persons_streaming_view_name: "persons_ab12_streaming"  # scratch name
+   ```
+
+   This creates `persons_ab12_raw`, `persons_distinct_ids_ab12_raw`, and a
+   validation view at the scratch name. The batch table keeps the canonical
+   name, so nothing reader-visible changes.
+
+2. **Point viaduck at the team.** Add a static destination for the team to
+   BOTH persons pipelines' env values in charts (an entry shape example is in
+   `argocd/viaduck/values/managed-warehouse-prod-us-persons.yaml`): the
+   persons pipeline writes `posthog.persons_ab12_raw`, the distinct-ids
+   pipeline writes `posthog.persons_distinct_ids_ab12_raw`. New destinations
+   initialize at the source head: only changes from this moment onward
+   stream.
+
+   Do NOT repoint the team's control-plane `persons_table_name` override to
+   the raw name as the cutover mechanism: the batch backfill resolves its
+   write target from the same override, so repointing redirects the batch
+   export into viaduck's table. (Discovery-based destination sourcing is a
+   post-fleet-migration step for the persons pipelines; see the charts PR
+   description for why it stays off during migration.)
+
+3. **Final batch top-up.** Run the team's daily partition once more so the
+   batch table holds everything up to the streaming cutover point. Expect a
+   small gap-or-overlap window either way; the view is idempotent under
+   overlap because the raw tables hold latest-per-key, and a gap only means
+   minutes of lag, not loss (the changelogs retain days).
+
+4. **Validate.** Compare the batch table against the scratch view:
+
+   ```sql
+   SELECT count(*) FROM posthog.persons_ab12;            -- batch
+   SELECT count(*) FROM posthog.persons_ab12_streaming;  -- streamed
+   ```
+
+   Counts should match within streaming lag. Spot-check a few recently
+   updated persons for property freshness.
+
+5. **Swap the view in.** Drop the batch table (or rename it to
+   `persons_ab12_batch` for a retention period) and re-run the prep config
+   with `persons_streaming_view_name: "persons_ab12"`. Readers keep the same
+   table name and columns.
+
+6. **Stop the batch top-up.** Disable warehouse backfill for the team (the
+   same control-plane enablement the sensors enumerate), so
+   `duckling_persons_daily_backfill_sensor` and the full-backfill sensor stop
+   creating partitions for it.
+
+## Rollback
+
+Until step 5, rollback is "do nothing": the batch table is untouched and the
+raw tables are inert extra objects. After step 5, rollback is: drop the view,
+rename `persons_ab12_batch` back to `persons_ab12`, re-enable backfill for
+the team, and run a full-export partition to close any gap. The raw tables
+can stay (viaduck keeps them current, harmless) or be dropped.
+
+## Open decisions
+
+- **View naming vs reader migration.** This runbook keeps the canonical name
+  on the view so readers never change. The alternative is leaving
+  `persons_ab12` on the raw table and migrating readers to a new view name,
+  which avoids the drop-and-swap in step 5 but breaks every saved query
+  against the batch shape.
+- **The viaduck rowid-reuse window.** A delete followed by a recreate of the
+  same key inside one viaduck flush window (~2 min) can drop the recreated
+  row (known full_cdc issue). Rare for persons, but a person deleted and
+  recreated quickly can briefly vanish from the view until the next version
+  arrives.
+- **Update fan-out cost.** A person property update rewrites one raw row but
+  changes N view rows (one per distinct_id). Views make this free at write
+  time and slightly more expensive at read time than the batch table. Measure
+  on the first high-traffic team before fleet cutover.
+
+## Cross-repo pieces
+
+| Repo | Piece | Status |
+|---|---|---|
+| posthog (this repo) | Streamed schema DDL + prep config flag + this runbook | this PR |
+| duckgres | `persons_distinct_ids_table` in the discovery payload | PostHog/duckgres#1026 |
+| viaduck | `discovery.table_field` to select the persons table fields | PostHog/viaduck#67 |
+| millpond | nothing: the `person` / `person-distinct-id` consumers already run in prod and land the changelogs on megaduck | done |
+| charts | viaduck persons pipeline ApplicationSets + values (disabled), `table_field` chart support | PostHog/charts#13846 |

posthog/dags/README_DUCKLINGS.md

@@ -2,6 +2,8 @@
 
 This document describes the Dagster jobs and sensors for backfilling ClickHouse data to customer-specific "ducklings" - isolated DuckLake instances with their own RDS catalog and S3 bucket.
 
+For the persons migration from this batch system to streaming replication (millpond + viaduck), see [PERSONS_STREAMING_MIGRATION.md](PERSONS_STREAMING_MIGRATION.md).
+
 ## Architecture
 
 ```text

posthog/dags/events_backfill_to_duckling.py

@@ -831,6 +831,98 @@ def _repair_stale_running_statuses(context: SensorEvaluationContext, dataset: st
 )
 """
 
+# --- Streamed persons schema (millpond/viaduck path) ---------------------------
+# The batch persons export is denormalized: one row per distinct_id with person
+# properties stamped on. Streaming replication (viaduck full_cdc out of the
+# shared millpond-landed changelogs) cannot produce that shape - it replicates
+# row-level changes keyed by (team_id, id) / (team_id, distinct_id) - so a
+# streamed duckling carries two raw current-state tables plus a view that
+# reproduces the batch shape for readers. Raw tables hold the LATEST version
+# per key (viaduck upserts by key) including is_deleted=1 tombstones, which the
+# source changelogs only ever emit as upserts; the view filters them.
+# See posthog/dags/PERSONS_STREAMING_MIGRATION.md for the full cutover runbook.
+
+# Same columns as the shared persons changelog (the clickhouse_person Kafka
+# record), keyed by (team_id, id). is_deleted is a real column here - unlike
+# the batch table, which filters it at export time.
+PERSONS_STREAMING_TABLE_DDL = """
+CREATE TABLE IF NOT EXISTS {catalog}.posthog.{table} (
+    team_id BIGINT,
+    id VARCHAR,
+    properties VARCHAR,
+    created_at TIMESTAMPTZ,
+    is_identified BOOLEAN,
+    is_deleted BOOLEAN,
+    version UBIGINT,
+    _timestamp TIMESTAMPTZ,
+    _inserted_at TIMESTAMPTZ
+)
+"""
+
+# Keyed by (team_id, distinct_id). Mirrors the person_distinct_id2 changelog.
+PERSONS_DISTINCT_IDS_TABLE_DDL = """
+CREATE TABLE IF NOT EXISTS {catalog}.posthog.{table} (
+    team_id BIGINT,
+    distinct_id VARCHAR,
+    person_id VARCHAR,
+    is_deleted BOOLEAN,
+    version BIGINT,
+    _timestamp TIMESTAMPTZ,
+    _inserted_at TIMESTAMPTZ
+)
+"""
+
+# Reader-compatible projection of the two raw tables: exactly the batch
+# persons table's columns (EXPECTED_DUCKLAKE_PERSONS_COLUMNS), latest state
+# only, tombstones filtered.
+PERSONS_DENORMALIZED_VIEW_DDL = """
+CREATE OR REPLACE VIEW {catalog}.posthog.{view} AS
+SELECT
+    p.team_id AS team_id,
+    pd.distinct_id AS distinct_id,
+    p.id AS id,
+    p.properties AS properties,
+    p.created_at AS created_at,
+    p.is_identified AS is_identified,
+    pd.version AS person_distinct_id_version,
+    p.version AS person_version,
+    p._timestamp AS _timestamp,
+    p._inserted_at AS _inserted_at
+FROM {catalog}.posthog.{persons_table} AS p
+INNER JOIN {catalog}.posthog.{distinct_ids_table} AS pd
+    ON p.id = pd.person_id AND p.team_id = pd.team_id
+WHERE p.is_deleted = FALSE AND pd.is_deleted = FALSE
+"""
+
+
+def persons_distinct_ids_table_name(persons_table: str) -> str:
+    """Derive a team's distinct-ids table from its persons table name.
+
+    Must stay rule-for-rule identical to the duckgres control plane's
+    derivation (distinctIDsTableName in controlplane/provisioning/discovery.go),
+    which serves this name to viaduck as persons_distinct_ids_table: a
+    `persons` prefix is replaced with `persons_distinct_ids` so a suffixed
+    team keeps its suffix (persons_ab12 -> persons_distinct_ids_ab12); any
+    other name gets `_distinct_ids` appended. If the two rules drift, viaduck
+    writes one table while this view joins another.
+    """
+    if persons_table.startswith("persons"):
+        return "persons_distinct_ids" + persons_table[len("persons") :]
+    return persons_table + "_distinct_ids"
+
+
+def persons_streaming_table_names(persons_table: str) -> tuple[str, str]:
+    """The (raw persons, distinct-ids) table names a streamed team's duckling uses.
+
+    The batch persons table keeps the canonical name until cutover, so the raw
+    streaming tables get `_raw` appended (persons_ab12 -> persons_ab12_raw).
+    At cutover the control-plane persons_table_name override is repointed to
+    the raw name, discovery then serves it (and the distinct-ids derivation
+    below) to viaduck, and the canonical name becomes the denormalized view.
+    """
+    raw = f"{persons_table}_raw"
+    return raw, persons_distinct_ids_table_name(raw)
+
 
 class DucklingBackfillConfig(Config):
     """Config for duckling events backfill job."""
@@ -852,6 +944,11 @@ class DucklingBackfillConfig(Config):
     # Huge team-days produce many right-sized files; tiny ones stay a single file.
     target_rows_per_file: int = TARGET_ROWS_PER_FILE
     max_s3_file_fanout: int = MAX_S3_FILE_FANOUT
+    # Persons streaming migration (posthog/dags/PERSONS_STREAMING_MIGRATION.md):
+    # create the raw streamed tables + denormalized view alongside the batch
+    # export. Persons-only; ignored by the events asset.
+    create_persons_streaming_schema: bool = False
+    persons_streaming_view_name: str | None = None  # defaults to <persons_table>_streaming
 
 
 def _events_row_group_buffer_fanout_limit(config: DucklingBackfillConfig) -> int:
@@ -1195,6 +1292,61 @@ def ensure_persons_table_exists(
     return True
 
 
+def ensure_persons_streaming_schema(
+    context: AssetExecutionContext,
+    target: DucklingTarget,
+    conn: psycopg.Connection[Any],
+    view_name: str,
+) -> None:
+    """Create the streamed-persons tables and reader view in the duckling catalog.
+
+    Prepares a duckling for viaduck-driven persons replication (see
+    posthog/dags/PERSONS_STREAMING_MIGRATION.md): the raw persons table
+    (is_deleted tombstones included), the distinct-ids table, and the
+    denormalized view reproducing the batch persons table's shape. The raw
+    tables take `_raw` names so the batch table keeps the canonical name
+    until cutover; `view_name` is explicit because the view's final name
+    (the canonical persons table name) is only free after the batch table
+    is dropped - run once pre-cutover with a scratch name to validate
+    against the batch table, then again at cutover with the canonical one.
+
+    Idempotent: safe to re-run and safe under concurrent partition runs.
+    """
+    alias = DUCKLAKE_ALIAS
+    persons_table, distinct_ids_table = persons_streaming_table_names(target.persons_table)
+    _validate_identifier(view_name)
+
+    context.log.info(f"Creating streamed persons tables ({persons_table}, {distinct_ids_table}) if missing...")
+    conn.execute(f"CREATE SCHEMA IF NOT EXISTS {alias}.posthog")
+    conn.execute(PERSONS_STREAMING_TABLE_DDL.format(catalog=alias, table=persons_table))
+    conn.execute(PERSONS_DISTINCT_IDS_TABLE_DDL.format(catalog=alias, table=distinct_ids_table))
+
+    # The streamed tables are written continuously by viaduck, not re-exported
+    # per day, so they partition on _inserted_at like the millpond-landed
+    # source tables do.
+    for table in (persons_table, distinct_ids_table):
+        _set_table_partitioning(conn, alias, table, "year(_inserted_at), month(_inserted_at)", context, target.team_id)
+
+    context.log.info(f"Creating denormalized persons view {view_name}...")
+    conn.execute(
+        PERSONS_DENORMALIZED_VIEW_DDL.format(
+            catalog=alias,
+            view=view_name,
+            persons_table=persons_table,
+            distinct_ids_table=distinct_ids_table,
+        )
+    )
+
+    logger.info(
+        "duckling_persons_streaming_schema_created",
+        team_id=target.team_id,
+        bucket=target.bucket,
+        persons_table=persons_table,
+        distinct_ids_table=distinct_ids_table,
+        view=view_name,
+    )
+
+
 def validate_duckling_schema(
     context: AssetExecutionContext,
     target: DucklingTarget,
@@ -2718,6 +2870,14 @@ def _run_duckling_persons_backfill(context: AssetExecutionContext, config: Duckl
                 context.log.info("Validating duckling persons schema compatibility...")
                 session.run("validate persons schema", lambda c: validate_duckling_persons_schema(context, target, c))
 
+            if config.create_persons_streaming_schema:
+                view_name = config.persons_streaming_view_name or f"{target.persons_table}_streaming"
+                context.log.info(f"Creating streamed persons schema (view {view_name}) in duckling catalog...")
+                session.run(
+                    "ensure persons streaming schema",
+                    lambda c: ensure_persons_streaming_schema(context, target, c, view_name),
+                )
+
         merged_settings = DEFAULT_CLICKHOUSE_SETTINGS.copy()
         merged_settings.update(settings_with_log_comment(context))
         if config.clickhouse_settings: