diff --git a/posthog/dags/PERSONS_STREAMING_MIGRATION.md b/posthog/dags/PERSONS_STREAMING_MIGRATION.md new file mode 100644 index 000000000000..6e9929bde6eb --- /dev/null +++ b/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__raw + persons_distinct_ids__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 | diff --git a/posthog/dags/README_DUCKLINGS.md b/posthog/dags/README_DUCKLINGS.md index 0823101d75e4..fe402fa493ce 100644 --- a/posthog/dags/README_DUCKLINGS.md +++ b/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 diff --git a/posthog/dags/events_backfill_to_duckling.py b/posthog/dags/events_backfill_to_duckling.py index 369635c9f83d..dfb07e63d889 100644 --- a/posthog/dags/events_backfill_to_duckling.py +++ b/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 _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: diff --git a/posthog/dags/test_events_backfill_to_duckling.py b/posthog/dags/test_events_backfill_to_duckling.py index a9985304c350..9afea23adca3 100644 --- a/posthog/dags/test_events_backfill_to_duckling.py +++ b/posthog/dags/test_events_backfill_to_duckling.py @@ -22,6 +22,9 @@ MAX_S3_FILE_FANOUT, PERSONS_COLUMNS, PERSONS_CONCURRENCY_TAG, + PERSONS_DENORMALIZED_VIEW_DDL, + PERSONS_DISTINCT_IDS_TABLE_DDL, + PERSONS_STREAMING_TABLE_DDL, PERSONS_TABLE_DDL, TARGET_ROWS_PER_FILE, DucklingBackfillConfig, @@ -56,6 +59,8 @@ is_full_export_partition, parse_partition_key, parse_partition_key_dates, + persons_distinct_ids_table_name, + persons_streaming_table_names, register_files_with_duckling, register_persons_files_with_duckling, table_exists, @@ -369,6 +374,61 @@ def test_persons_ddl_honors_suffixed_table_name(self): conn.close() +class TestPersonsStreamingSchema: + # The derivation must stay rule-for-rule identical to the duckgres control + # plane's distinctIDsTableName: drift means viaduck writes one table while + # the denormalized view joins another. + @parameterized.expand( + [ + ("default", "persons", "persons_distinct_ids"), + ("suffixed", "persons_ab12", "persons_distinct_ids_ab12"), + ("non_prefixed_override", "customers", "customers_distinct_ids"), + ] + ) + def test_distinct_ids_table_name_derivation(self, _label, persons_table, expected): + assert persons_distinct_ids_table_name(persons_table) == expected + + def test_streaming_table_names_keep_batch_name_free(self): + raw, distinct_ids = persons_streaming_table_names("persons_ab12") + assert (raw, distinct_ids) == ("persons_ab12_raw", "persons_distinct_ids_ab12_raw") + + def test_view_reproduces_batch_shape_and_filters_tombstones(self): + conn = duckdb.connect() + conn.execute("CREATE SCHEMA IF NOT EXISTS memory.posthog") + raw, distinct_ids = persons_streaming_table_names("persons") + conn.execute(PERSONS_STREAMING_TABLE_DDL.format(catalog="memory", table=raw)) + conn.execute(PERSONS_DISTINCT_IDS_TABLE_DDL.format(catalog="memory", table=distinct_ids)) + conn.execute( + PERSONS_DENORMALIZED_VIEW_DDL.format( + catalog="memory", + view="persons_streaming", + persons_table=raw, + distinct_ids_table=distinct_ids, + ) + ) + + conn.execute( + f"INSERT INTO memory.posthog.{raw} VALUES " + "(2, 'person-live', '{}', now(), true, false, 3, now(), now())," + "(2, 'person-gone', '{}', now(), true, true, 4, now(), now())" + ) + conn.execute( + f"INSERT INTO memory.posthog.{distinct_ids} VALUES " + "(2, 'did-live', 'person-live', false, 1, now(), now())," + "(2, 'did-gone', 'person-gone', false, 1, now(), now())," + "(2, 'did-detached', 'person-live', true, 2, now(), now())" + ) + + columns = {row[0] for row in conn.execute("DESCRIBE memory.posthog.persons_streaming").fetchall()} + assert columns == EXPECTED_DUCKLAKE_PERSONS_COLUMNS + + rows = conn.execute("SELECT distinct_id, id FROM memory.posthog.persons_streaming").fetchall() + # The deleted person and the deleted mapping are filtered; the live + # person's remaining mapping still projects. + assert rows == [("did-live", "person-live")] + conn.close() + + class TestParsePartitionKeyDates: @patch("posthog.dags.events_backfill_to_duckling.timezone") def test_daily_format_returns_single_date(self, mock_timezone):