From cf6d47415f6485e8ff2a3a186d05c1554a9ba3ca Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Wed, 12 Aug 2026 11:04:11 -0700 Subject: [PATCH 1/2] feat(flags): sweep flag_evaluations in data deletions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit flag_evaluations stores the same person payload as events — the full properties JSON, person_properties, person_id and distinct_id — but no deletion scanner reached it, leaving its TTL as the only erasure control. The table is empty everywhere and has no producer, so this is a gate before one lands rather than a live fix. Which tables a deletion must sweep is a property of the data they store, not of the events table, so posthog/models/deletion_targets.py now owns that list and posthog/models/event/deletion.py becomes a shim over it. Per-target capability fields carry what each table's schema can take: which HogQL schema a predicate compiles against, whether its property columns can be rewritten in place, which events it can hold. Person deletion, team deletion, the queued-uuid drain and event removal all sweep the new table, keyed on person_id like every other events-shaped table. That relies on the producer populating person_id as the events pipeline does — the resolved uuid, else a deterministic per-distinct_id uuid — a contract recorded in docs/internal/clickhouse-deletion-coverage.md. Property removal and HogQL-predicate event removal cannot reach the table: its nine typed columns are true ClickHouse MATERIALIZED, so ClickHouse rejects both an UPDATE of them and of properties itself, and no HogQL table definition exists for it. Both refuse loudly when the table holds matching rows rather than completing while data survives. The refusal check is bounded by the request's deletion marker, so a row ingested after the marker — one the sweep itself would never touch — can't refuse the request forever. find_existing_mutations, which every ClickHouse mutation in the repo goes through for dedup, interpolated the command template into a $__sql$ heredoc and let the driver substitute parameters inside it. The driver escapes quotes and backslashes but not `$`, so a parameter value containing the delimiter closed the heredoc early and the remainder parsed as SQL, silently, since the surrounding array kept its length. Mutation parameters were staff-authored until this branch's person sweep put a person's distinct_id there, so that is reachable input now. Each command is rendered in Python and bound as an ordinary parameter instead, and a new test pins a delimiter-shaped parameter value against exactly this construction. A coverage test fails on any storage table that declares person_properties and is neither registered nor listed as TTL-only, so the next events-shaped table forces the same decision. The tables that rely on TTL alone erase later than their TTL suggests: ttl_only_drop_parts means a part drops only once its newest row expires, so the real wait is ~8 days for sharded_events_recent (daily partitions) and ~120 days for flag_evaluations (monthly), documented in docs/internal/clickhouse-deletion-coverage.md alongside the sweep table. Fixes #81002 Generated-By: PostHog Code Task-Id: b6e91f38-dd91-4c39-bfce-322d583c221e --- docs/internal/clickhouse-deletion-coverage.md | 103 ++++++++ posthog/clickhouse/cluster.py | 22 +- posthog/clickhouse/test/test_cluster.py | 24 ++ .../clickhouse/test/test_deletion_coverage.py | 45 ++++ posthog/dags/data_deletion_requests.py | 210 +++++++++++++--- posthog/dags/deletes.py | 10 +- posthog/dags/tests/conftest.py | 11 + .../dags/tests/test_data_deletion_requests.py | 148 +++++++++++ posthog/dags/tests/test_deletes.py | 52 ++++ posthog/models/data_deletion_request.py | 50 +++- posthog/models/deletion_targets.py | 237 ++++++++++++++++++ posthog/models/event/deletion.py | 61 ++--- posthog/models/flag_evaluations/sql.py | 25 ++ posthog/test/base.py | 12 + 14 files changed, 909 insertions(+), 101 deletions(-) create mode 100644 docs/internal/clickhouse-deletion-coverage.md create mode 100644 posthog/clickhouse/test/test_deletion_coverage.py create mode 100644 posthog/models/deletion_targets.py diff --git a/docs/internal/clickhouse-deletion-coverage.md b/docs/internal/clickhouse-deletion-coverage.md new file mode 100644 index 000000000000..624efcdfa981 --- /dev/null +++ b/docs/internal/clickhouse-deletion-coverage.md @@ -0,0 +1,103 @@ +# ClickHouse deletion coverage + +Which ClickHouse tables a data deletion reaches, which rely on their TTL instead, and why. + +Deleting a person's data is not a property of the events table. +It is a property of every table that stores rows attributable to a person. +`posthog/models/deletion_targets.py` is the one list of those tables; this document is the reasoning behind it. + +## The sweeps + +| Sweep | Entry point | Predicate columns | +| ------------------------ | -------------------------------- | ------------------------------------------ | +| Person deletion (async) | `deletes_job` → `delete_events` | `team_id`, `person_id`, `timestamp` | +| Team deletion | `deletes_job` → `delete_events` | `team_id` | +| Queued uuid drain | `deletes_job` → `delete_events` | `team_id`, `uuid` | +| Person removal request | `delete_person_events_op` | `team_id`, `person_id`, `timestamp` | +| Event removal request | `execute_event_deletion` | `team_id`, `timestamp`, `event`, + HogQL | +| Property removal request | `process_property_removal_shard` | `properties`, `person_properties`, + HogQL | + +The first four use only columns every target declares, so they apply unchanged to any registered table. +The last two need more, which is what the capability fields on `DeletionTarget` express. + +Team deletion for tables that are replicated rather than sharded runs through a separate per-table loop (`delete_team_data_from`), which dispatches to a single host. +A sharded table must not be registered there — it would sweep one shard. +The team arm inside `delete_events` covers sharded tables. + +## Covered tables + +- `sharded_events` — all sweeps. +- `sharded_events_json` — all sweeps. Optional: only present after the native-JSON migration. +- `sharded_flag_evaluations` — person, team, queued-uuid and event removal. Not property removal (below). Optional. + +## Tables on TTL alone + +Listed in `TTL_ONLY_TABLES`. +Each is a decision that erasure may lag by the retention window. + +- `sharded_events_recent` — a transient mirror of the last few days of events, 7-day TTL keyed on `inserted_at`. It partitions by day with `ttl_only_drop_parts = 1`, so a part drops only once its newest row expires: the real worst case is about 8 days plus TTL-merge lag, not a flat 7. Short enough to accept as the erasure bound, and a sweep would race the TTL for little benefit. + +Session recordings, the dead letter queue, and logs are likewise TTL-reclaimed. +That decision predates this document; the older `posthog/models/async_deletion/delete_events.py` records it in a comment, but that module is legacy and is not the source of truth here. + +## Known gaps + +### Property removal does not reach `flag_evaluations` + +The events property-removal path rewrites rows in a staging table and resets each affected materialized column with `ALTER TABLE … UPDATE = ''`. +That works because `materialize()` creates columns as `DEFAULT `, which is assignable. + +`flag_evaluations` declares its nine typed columns as true ClickHouse `MATERIALIZED`. Measured against ClickHouse 26.6: + +- Assigning to one is rejected: `Cannot UPDATE materialized column 'session_id'`. +- Updating `properties` is rejected too, because `flag_key` is materialized from it and sits in the sort key: `Updated column 'properties' affects MATERIALIZED column 'flag_key', which is a key column`. +- `CREATE TABLE tmp AS sharded_flag_evaluations ENGINE = MergeTree()` inherits the sort key and the column kinds, so the staging table hits the same rejection. + +Rewriting these rows therefore needs a different algorithm: +`INSERT … SELECT` the cleaned rows and let the shard recompute the typed columns on write, then lightweight-delete the originals. +Until that exists, `get_property_removal_shards` refuses to start when the table holds rows matching the request, so a request cannot complete while data it named survives. +The check costs nothing while the table is empty. + +This is also why `flag_evaluations` is deliberately absent from `MATERIALIZATION_VALID_TABLES`. +Adding it would let `materialize()` mint `DEFAULT`-kind columns on a table whose property-removal path cannot reset them. + +#### Decide the escape hatch before a producer lands + +Today the refusal costs nothing, because the table is empty. +Once it holds rows, a property removal with `delete_all_events` refuses whenever a single flag-evaluation row carries the named property, and the operator has no way through: `delete_all_events` and `events` are mutually exclusive on the model, so the request cannot be narrowed to exclude `$feature_flag_called`, and the admin Retry button replays the same failure. +The only exits are waiting out the TTL or shipping the re-insert algorithm. The table partitions by month with `ttl_only_drop_parts = 1`, so a part drops only once its newest row expires: the real wait is up to about 120 days, not the 90-day TTL. `posthog/models/flag_evaluations/sql.py` says the same thing next to the partition clause. + +That is the right default (refusing beats silently under-deleting), but it needs a deliberate answer before real traffic hits it. The options, in rough order of cost: + +1. Ship the re-insert rewrite, which removes the refusal entirely. +2. Let a request exclude event names, so an operator can scope around the table. +3. Record an explicit, audited acknowledgement on the request so an operator can accept the residue rather than being stuck. + +Doing nothing means the first affected GDPR request becomes an escalation. + +### Event removal with a HogQL predicate does not reach `flag_evaluations` + +`compile_hogql_predicate` resolves against the events HogQL table and emits events-specific physical columns. There is no HogQL table definition for `flag_evaluations`, so the fragment cannot run against it. Requests without a predicate are swept normally; requests with one are refused if the table holds matching rows. + +## Producer prerequisite: person_id parity + +Every person sweep is keyed on `person_id`, on `flag_evaluations` as on events. +That is only correct because the producer populates `person_id` for every row, exactly as the events ingestion pipeline does: the resolved `Person.uuid`, or a deterministic per-`distinct_id` uuid when no person was resolved. +The shadow-routing producer (`posthog-code/flag-evaluations-shadow-routing`) forks the enriched event downstream of person resolution, so it inherits that value for free. + +If a future producer emitted rows before person resolution, leaving `person_id` unset, a person deletion would strand them. +The fix would belong to the producer, not the scanner. +Keeping the fork downstream of person resolution is the contract, tracked on #81002. + +## Related, and deliberately unchanged + +`_fetch_stats` counts only the events tables. It feeds `AUTO_APPROVE_MAX_EVENTS`, a cost heuristic rather than a completeness claim, so a request auto-approved as small may move somewhat more rows than measured. + +`cleanup_old_events_by_partition` stays events-only. It enforces a multi-year retention floor for a named set of teams, and every other personal-data table already expires sooner under its own TTL. + +## Adding a table + +Register it in `PERSONAL_DATA_TARGETS`, with capability flags reflecting what its schema can actually take. +If it is not going to be swept, add it to `TTL_ONLY_TABLES` with the window you are accepting. + +`posthog/clickhouse/test/test_deletion_coverage.py` fails on any storage table that declares `person_properties` and appears in neither list, so the decision has to be made rather than skipped. diff --git a/posthog/clickhouse/cluster.py b/posthog/clickhouse/cluster.py index 510f51cfe91a..6be3f3a61c39 100644 --- a/posthog/clickhouse/cluster.py +++ b/posthog/clickhouse/cluster.py @@ -856,7 +856,17 @@ def find_existing_mutations(self, client: Client, commands: Set[str] | None = No # ClickHouse normalizes bare references against the connection database when # storing the mutation, and a bare-vs-qualified mismatch defeats the join. alter_prefix = f"ALTER TABLE {settings.CLICKHOUSE_DATABASE}.{self.table} " - per_command_alters = ", ".join(f"$__sql${alter_prefix}{cmd}$__sql$" for cmd in command_list) + # Render each command's parameters here and bind the finished text as an ordinary parameter, + # rather than interpolating the template into a $__sql$ heredoc and letting the driver + # substitute values inside it. The driver escapes quotes and backslashes but not `$`, so a + # value containing the heredoc delimiter would close it early and the rest would parse as + # SQL. Mutation parameters carry third-party strings (a person's distinct_id), so that is + # reachable input, and the injection is silent because the surrounding array keeps its length. + rendered_commands = [ + client.substitute_params(f"{alter_prefix}{cmd}", self.parameters, client.connection.context) + for cmd in command_list + ] + per_command_alters = ", ".join(f"%(__command_{i})s" for i in range(len(rendered_commands))) mutations = client.execute( f""" SELECT mutation_id @@ -893,10 +903,12 @@ def find_existing_mutations(self, client: Client, commands: Set[str] | None = No SETTINGS join_use_nulls = 1 """, { - f"__database": settings.CLICKHOUSE_DATABASE, - f"__table": self.table, - f"__alter_prefix": alter_prefix, - **self.parameters, + "__database": settings.CLICKHOUSE_DATABASE, + "__table": self.table, + "__alter_prefix": alter_prefix, + # self.parameters are already rendered into __command_*; passing them again would + # reintroduce the substitution this avoids. + **{f"__command_{i}": text for i, text in enumerate(rendered_commands)}, }, ) assert len(mutations) == len(command_list) diff --git a/posthog/clickhouse/test/test_cluster.py b/posthog/clickhouse/test/test_cluster.py index 1dcd67dbae5e..14be3de94f01 100644 --- a/posthog/clickhouse/test/test_cluster.py +++ b/posthog/clickhouse/test/test_cluster.py @@ -339,6 +339,30 @@ def test_find_existing_mutations_handles_multiline_formatted_command(cluster: Cl assert shard_mutations == duplicate +def test_find_existing_mutations_handles_delimiter_shaped_parameter_value(cluster: ClickhouseCluster) -> None: + """Regression test: `find_existing_mutations` used to interpolate the command template into a + `$__sql$` heredoc and let the driver substitute parameters inside it. The driver escapes quotes + and backslashes but not `$`, so a parameter value containing the delimiter closed the heredoc + early and the remainder of the value parsed as SQL — silently, since the surrounding array kept + its length. A person's `distinct_id` is exactly such attacker-controlled input once it reaches a + mutation predicate. Parameters are now rendered via `substitute_params` and bound as an ordinary + parameter, so a delimiter-shaped value is just data; pre-fix, this raised a ClickHouse syntax + error instead of returning cleanly. + """ + sentinel_uuid = uuid.uuid1() # keeps the rendered command unique to this run + + runner = AlterTableMutationRunner( + table=EVENTS_DATA_TABLE(), + commands={"UPDATE person_id = %(uuid)s WHERE distinct_id = %(distinct_id)s"}, + parameters={"uuid": sentinel_uuid, "distinct_id": "$__sql$"}, + ) + + existing = cluster.map_all_hosts(runner.find_existing_mutations).result() + assert all(not mutations for mutations in existing.values()), ( + "expected no pre-existing mutation for this delimiter-shaped parameter" + ) + + def test_alter_mutation_multiple_commands(cluster: ClickhouseCluster) -> None: table = EVENTS_DATA_TABLE() count = 100 diff --git a/posthog/clickhouse/test/test_deletion_coverage.py b/posthog/clickhouse/test/test_deletion_coverage.py new file mode 100644 index 000000000000..93a663cab711 --- /dev/null +++ b/posthog/clickhouse/test/test_deletion_coverage.py @@ -0,0 +1,45 @@ +from posthog.test.base import BaseTest, ClickhouseTestMixin + +from django.conf import settings + +from posthog.clickhouse.client import sync_execute +from posthog.models.deletion_targets import PERSONAL_DATA_TARGETS, TTL_ONLY_TABLES + + +class TestDeletionCoverage(ClickhouseTestMixin, BaseTest): + def test_storage_tables_carrying_person_properties_are_swept_or_exempt(self): + # A table that stores person_properties stores personal data, so deleting a person has to + # reach it. flag_evaluations shipped without any scanner pointed at it precisely because + # nothing forced that decision to be made out loud (#81002). + # + # Keyed on person_properties rather than person_id: person_id is on cohortpeople, + # person_static_cohort, person_distinct_id2 and more, which makes for an exemption list too + # long to carry any signal. person_properties is the events-mirror signature. + # + # tmp_ is excluded because the property-removal job's staging tables are created AS + # sharded_events, so a leaked one would fail this with a misleading name. + rows = sync_execute( + """ + SELECT t.name + FROM system.tables t + INNER JOIN system.columns c ON c.database = t.database AND c.table = t.name + WHERE t.database = %(database)s + AND c.name = 'person_properties' + AND t.engine LIKE '%%MergeTree%%' + AND t.name NOT LIKE 'tmp\\_%%' + ORDER BY t.name + """, + {"database": settings.CLICKHOUSE_DATABASE}, + ) + storage_tables = {row[0] for row in rows} + assert storage_tables, "expected at least the events tables to declare person_properties" + + accounted_for = {target.data_table for target in PERSONAL_DATA_TARGETS} | TTL_ONLY_TABLES + unaccounted = sorted(storage_tables - accounted_for) + + assert not unaccounted, ( + f"{unaccounted} store person_properties but no deletion sweep reaches them. Register each " + "in PERSONAL_DATA_TARGETS (posthog/models/deletion_targets.py), or add it to " + "TTL_ONLY_TABLES with the retention window you are accepting as its erasure bound. " + "See docs/internal/clickhouse-deletion-coverage.md." + ) diff --git a/posthog/dags/data_deletion_requests.py b/posthog/dags/data_deletion_requests.py index fbdef5826de2..1298545c779d 100644 --- a/posthog/dags/data_deletion_requests.py +++ b/posthog/dags/data_deletion_requests.py @@ -37,9 +37,16 @@ event_match_sql_fragment, event_removal_where, jsonhas_expr, + portable_event_removal_where, verify_queued_request, ) -from posthog.models.event.deletion import events_data_tables +from posthog.models.deletion_targets import ( + DeletionTarget, + UnsweepableRowsError, + assert_no_unsweepable_rows, + resolve_targets, +) +from posthog.models.event.deletion import cluster_has_events_json_table from posthog.models.event.sql import ( DISTRIBUTED_EVENTS_JSON_TABLE, EVENTS_DATA_TABLE, @@ -299,6 +306,9 @@ def _get_affected_mat_columns( filter by ``default_kind`` on the same row that carries the comment. The comment itself is a sufficient identifier — it is PostHog-specific and the ``elements_chain::*`` family is excluded explicitly. + + Callers pass ``"events"`` deliberately; see docs/internal/clickhouse-deletion-coverage.md for + why the other tables' typed columns cannot be discovered or reset here. """ database = django_settings.CLICKHOUSE_DATABASE sql = """ @@ -419,29 +429,82 @@ def load_deletion_request( ) +_COVERAGE_DOC = "docs/internal/clickhouse-deletion-coverage.md" + +_HOGQL_UNSWEEPABLE_REASON = ( + "the request carries a HogQL predicate, which only compiles against the events schema " + "(this table has no HogQL table definition, so the compiled fragment names columns it lacks). " + "To proceed, re-file the request without the predicate, or narrow its events to ones this " + f"table never stores. See {_COVERAGE_DOC}." +) + + +def _refuse_unsweepable( + cluster: ClickhouseCluster, + targets: list[DeletionTarget], + deletion_request: DeletionRequestContext, + predicate: str, + params: dict, + *, + reason: str, +) -> None: + """Raise a dagster.Failure when the request would strand rows on any of ``targets``.""" + try: + assert_no_unsweepable_rows( + cluster, + targets, + predicate, + params, + events=[] if deletion_request.delete_all_events else deletion_request.events, + reason=reason, + ) + except UnsweepableRowsError as exc: + raise dagster.Failure(description=f"Deletion request {deletion_request.request_id}: {exc}") from exc + + +def _event_removal_targets( + cluster: ClickhouseCluster, deletion_request: DeletionRequestContext +) -> list[DeletionTarget]: + """Targets this event-removal request can actually sweep, refusing to strand rows on the rest.""" + events = [] if deletion_request.delete_all_events else deletion_request.events + # A target that can't hold any of the named events has nothing to sweep, and mutations serialize + # per table, so enqueueing a no-op one would queue in front of real work. + targets = [target for target in resolve_targets(cluster) if target.may_hold_any_of(events)] + if not deletion_request.hogql_predicate: + return targets + + unsweepable = [target for target in targets if not target.accepts_hogql_predicate] + if unsweepable: + predicate, params = portable_event_removal_where(deletion_request) + _refuse_unsweepable(cluster, unsweepable, deletion_request, predicate, params, reason=_HOGQL_UNSWEEPABLE_REASON) + return [target for target in targets if target.accepts_hogql_predicate] + + def _run_immediate_event_deletion( context: dagster.OpExecutionContext, cluster: ClickhouseCluster, deletion_request: DeletionRequestContext, ) -> None: - tables = events_data_tables(cluster) + targets = _event_removal_targets(cluster, deletion_request) shards = sorted(cluster.shards) - context.log.info(f"Starting immediate event deletion across {len(shards)} shards on tables {tables}") + context.log.info( + f"Starting immediate event deletion across {len(shards)} shards on tables {[t.data_table for t in targets]}" + ) - for table in tables: + for target in targets: # The HogQL fragment compiles differently per schema: materialized-column/JSONExtract # reads on the legacy table, JSON subcolumn reads on the native-JSON table. predicate, parameters = event_removal_where( - deletion_request, use_new_events_schema=table == EVENTS_JSON_DATA_TABLE + deletion_request, use_new_events_schema=target.uses_new_events_schema ) for idx, shard_num in enumerate(shards, 1): - context.log.info(f"Processing {table} shard {shard_num} ({idx}/{len(shards)})") + context.log.info(f"Processing {target.data_table} shard {shard_num} ({idx}/{len(shards)})") shard_start = time.monotonic() runner = LightweightDeleteMutationRunner( - table=table, + table=target.data_table, predicate=predicate, parameters=parameters, settings={"lightweight_deletes_sync": 0}, @@ -452,10 +515,14 @@ def _run_immediate_event_deletion( cluster.map_all_hosts_in_shard(shard_num, mutation_waiter.wait).result() elapsed = time.monotonic() - shard_start - context.log.info(f"{table} shard {shard_num} complete in {elapsed:.1f}s") + context.log.info(f"{target.data_table} shard {shard_num} complete in {elapsed:.1f}s") context.add_output_metadata( - {"mode": dagster.MetadataValue.text("immediate"), "shards_processed": dagster.MetadataValue.int(len(shards))} + { + "mode": dagster.MetadataValue.text("immediate"), + "shards_processed": dagster.MetadataValue.int(len(shards)), + "swept_tables": dagster.MetadataValue.text(", ".join(t.data_table for t in targets)), + } ) @@ -464,21 +531,24 @@ def _queue_events_for_deferred_deletion( cluster: ClickhouseCluster, deletion_request: DeletionRequestContext, ) -> None: - # Reading candidates from the legacy table only is fine: the queue holds (team_id, uuid) - # pairs and event UUIDs are identical across the legacy and native-JSON tables, so the - # deletes_job drain applies them to both. - source_table = EVENTS_DATA_TABLE() + # The queue holds (team_id, uuid) pairs, and the deletes_job drain applies them to every + # personal-data table. Flag-evaluation rows still have to be read on their own: they mirror a + # subset of events, so a uuid there may not be in sharded_events once routing moves those + # events off it. + sources = [t for t in _event_removal_targets(cluster, deletion_request) if t.queue_uuid_candidates] db = django_settings.CLICKHOUSE_DATABASE shards = sorted(cluster.shards) predicate, params = event_removal_where(deletion_request) - # nosemgrep: clickhouse-fstring-param-audit (all interpolated values are internal constants/settings) - insert_sql = ( - f"INSERT INTO {db}.{ADHOC_EVENTS_DELETION_TABLE} (team_id, uuid) " - f"SELECT team_id, uuid FROM {db}.{source_table} WHERE {predicate}" - ) def run_on_shard(client: Client) -> int: - client.execute(insert_sql, params, settings={"max_execution_time": 1800}) + for source in sources: + # nosemgrep: clickhouse-fstring-param-audit (all interpolated values are internal constants/settings) + client.execute( + f"INSERT INTO {db}.{ADHOC_EVENTS_DELETION_TABLE} (team_id, uuid) " + f"SELECT team_id, uuid FROM {db}.{source.data_table} WHERE {predicate}", + params, + settings={"max_execution_time": 1800}, + ) row = client.execute( f"SELECT count() FROM {db}.{ADHOC_EVENTS_DELETION_TABLE} WHERE team_id = %(team_id)s AND is_deleted = 0", {"team_id": params["team_id"]}, @@ -498,7 +568,11 @@ def run_on_shard(client: Client) -> int: context.log.info(f"Shard {shard_num}: queued ~{queued} rows in {elapsed:.1f}s") context.add_output_metadata( - {"mode": dagster.MetadataValue.text("deferred"), "queued_rows": dagster.MetadataValue.int(total_queued)} + { + "mode": dagster.MetadataValue.text("deferred"), + "queued_rows": dagster.MetadataValue.int(total_queued), + "queued_from": dagster.MetadataValue.text(", ".join(s.data_table for s in sources)), + } ) @@ -521,6 +595,15 @@ def execute_event_deletion( # --------------------------------------------------------------------------- +_PROPERTY_REWRITE_UNSWEEPABLE_REASON = ( + "its typed columns are MATERIALIZED from properties, so ClickHouse rejects both an assignment " + "to them and an update of properties itself; rewriting these rows needs a re-insert instead of " + "the staging-table mutation this job runs. There is no way to complete this request today: " + "either narrow its events to ones this table never stores, or wait out the table's TTL. " + f"See {_COVERAGE_DOC}." +) + + @dagster.op(tags=OWNER_TAG) def load_property_removal_request( context: dagster.OpExecutionContext, @@ -607,7 +690,31 @@ def get_property_removal_shards( Takes the deletion request as input so fan-out is sequenced after the load op; the mapping key makes each shard re-executable individually from the Dagster UI. + + Also the gate for targets this job cannot rewrite: refusing here, before any shard mutates + anything, is what stops the request completing while matching rows survive elsewhere. It lives + in this op rather than the load op because this is the first one holding a cluster handle. """ + unsweepable = [t for t in resolve_targets(cluster) if not t.accepts_property_rewrite] + if unsweepable: + marker = deletion_request.inserted_at_marker + if marker is None: + raise dagster.Failure( + description="property_removal_marker missing; load_property_removal_request must set it" + ) + marker_str = marker.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%S.%f") + # Bound by the same marker as the sweep and the verify gate, so a row ingested after the + # marker — which the sweep would never touch — can't refuse the request forever. + presence, presence_params = _property_removal_where(deletion_request, inserted_at_max=marker_str) + _refuse_unsweepable( + cluster, + unsweepable, + deletion_request, + presence, + presence_params, + reason=_PROPERTY_REWRITE_UNSWEEPABLE_REASON, + ) + shards = sorted(cluster.shards) context.log.info(f"Fanning out property removal {deletion_request.request_id} to {len(shards)} shard op(s)") for shard_num in shards: @@ -672,7 +779,7 @@ def process_property_removal_shard( targets: list[tuple[str, str, bool, tuple[str, dict]]] = [ (EVENTS_DATA_TABLE(), base_temp, False, compile_hogql_predicate(deletion_request)), ] - if EVENTS_JSON_DATA_TABLE in events_data_tables(cluster): + if cluster_has_events_json_table(cluster): targets.append( ( EVENTS_JSON_DATA_TABLE, @@ -924,12 +1031,28 @@ def verify_property_removal( if marker is None: raise dagster.Failure(description="property_removal_marker missing; load_property_removal_request must set it") marker_str = marker.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%S.%f") + + # Repeat the fan-out gate here. That one is point-in-time: rows can land between it and now, and + # a re-execution from a failed shard reuses the fan-out op's cached output without re-running it. + # Bounded by the same marker as the checks below so post-marker ingestion can't wedge the run. + unsweepable = [t for t in resolve_targets(cluster) if not t.accepts_property_rewrite] + if unsweepable: + presence, presence_params = _property_removal_where(deletion_request, inserted_at_max=marker_str) + _refuse_unsweepable( + cluster, + unsweepable, + deletion_request, + presence, + presence_params, + reason=_PROPERTY_REWRITE_UNSWEEPABLE_REASON, + ) + properties = deletion_request.properties person_properties = deletion_request.person_properties targets: list[tuple[str, bool, tuple[str, dict]]] = [ ("events", False, compile_hogql_predicate(deletion_request)), ] - if EVENTS_JSON_DATA_TABLE in events_data_tables(cluster): + if cluster_has_events_json_table(cluster): targets.append( ( DISTRIBUTED_EVENTS_JSON_TABLE, @@ -1088,7 +1211,13 @@ def load_person_removal_request( def _person_event_predicate(ctx: PersonRemovalContext) -> tuple[str, dict]: - """Build WHERE predicate + params for events linked to the targeted persons.""" + """Build WHERE predicate + params for rows linked to the targeted persons. + + Keyed on ``person_id`` only, like every other events-shaped deletion. The producers of these + tables populate ``person_id`` for every row (the resolved uuid, else a deterministic + per-distinct_id uuid), so a distinct_id arm would only widen the match to rows the person + already owns. + """ parts = ["team_id = %(team_id)s AND person_id IN %(person_ids)s"] params: dict = {"team_id": ctx.team_id, "person_ids": ctx.person_uuids} if ctx.start_time is not None and ctx.end_time is not None: @@ -1104,12 +1233,13 @@ def delete_person_events_op( cluster: dagster.ResourceParam[ClickhouseCluster], person_removal: PersonRemovalContext, ) -> PersonRemovalContext: - """Per-shard lightweight delete of events for the targeted persons.""" + """Per-shard lightweight delete of person-linked rows on every personal-data table.""" if not person_removal.drop_events: context.log.info("drop_events=False, skipping event deletion") return person_removal - # The CH events table is keyed by person_id (UUID), so resolve distinct_ids → uuids when - # the request was submitted by distinct_id. Selectors are mutually exclusive (enforced in + + # The tables are keyed by person_id (UUID), so resolve distinct_ids → uuids when the request was + # submitted by distinct_id. Selectors are mutually exclusive (enforced in # DataDeletionRequest._clean_person_removal and re-checked in load_person_removal_request). if person_removal.person_distinct_ids: persons = resolve_persons_for_deletion( @@ -1122,19 +1252,22 @@ def delete_person_events_op( context.log.info("No persons resolved; nothing to delete") return person_removal - # The predicate only references schema-agnostic columns (team_id, person_id, timestamp), so - # the same delete applies to both the legacy and native-JSON events tables. - tables = events_data_tables(cluster) - predicate, params = _person_event_predicate(person_removal) + targets = resolve_targets(cluster) shards = sorted(cluster.shards) - context.log.info(f"Deleting events for {len(person_removal.person_uuids)} persons across {len(shards)} shards") + context.log.info( + f"Deleting rows for {len(person_removal.person_uuids)} persons across {len(shards)} shards " + f"on tables {[t.data_table for t in targets]}" + ) - for table in tables: + # Schema-agnostic columns only (team_id, person_id, timestamp), so one predicate serves every + # target. + predicate, params = _person_event_predicate(person_removal) + for target in targets: for idx, shard_num in enumerate(shards, 1): - context.log.info(f"Processing {table} shard {shard_num} ({idx}/{len(shards)})") + context.log.info(f"Processing {target.data_table} shard {shard_num} ({idx}/{len(shards)})") shard_start = time.monotonic() runner = LightweightDeleteMutationRunner( - table=table, + table=target.data_table, predicate=predicate, parameters=params, settings={"lightweight_deletes_sync": 0}, @@ -1142,9 +1275,14 @@ def delete_person_events_op( shard_result = cluster.map_any_host_in_shards({shard_num: runner}).result() _host, waiter = next(iter(shard_result.items())) cluster.map_all_hosts_in_shard(shard_num, waiter.wait).result() - context.log.info(f"{table} shard {shard_num} complete in {time.monotonic() - shard_start:.1f}s") + context.log.info(f"{target.data_table} shard {shard_num} complete in {time.monotonic() - shard_start:.1f}s") - context.add_output_metadata({"shards_processed": dagster.MetadataValue.int(len(shards))}) + context.add_output_metadata( + { + "shards_processed": dagster.MetadataValue.int(len(shards)), + "swept_tables": dagster.MetadataValue.text(", ".join(t.data_table for t in targets)), + } + ) return person_removal diff --git a/posthog/dags/deletes.py b/posthog/dags/deletes.py index 05e37e32743e..b07903a94927 100644 --- a/posthog/dags/deletes.py +++ b/posthog/dags/deletes.py @@ -28,6 +28,7 @@ from posthog.dags.common import JobOwners from posthog.dags.person_overrides import squash_person_overrides from posthog.models.async_deletion import AsyncDeletion, DeletionType +from posthog.models.deletion_targets import personal_data_tables from posthog.models.event.deletion import events_data_tables from posthog.models.event.sql import EVENTS_DATA_TABLE from posthog.models.group.sql import GROUPS_TABLE @@ -601,8 +602,8 @@ def count_pending_adhoc_deletes(client: Client) -> int: } ) - # The same delete must land on every physical events table (legacy + native-JSON), or the - # tables diverge while both exist. + # The predicate below reads only team_id, person_id, timestamp and uuid — columns every + # registered target declares — so it applies unchanged to all of them. delete_mutation_runners = [ LightweightDeleteMutationRunner( table=table, @@ -619,7 +620,7 @@ def count_pending_adhoc_deletes(client: Client) -> int: "adhoc_event_deletes_dictionary": load_and_verify_adhoc_event_deletes_dictionary.qualified_name, }, ) - for table in events_data_tables(cluster) + for table in personal_data_tables(cluster) ] shard_waiters: dict[int, list[MutationWaiter]] = {} @@ -890,6 +891,9 @@ def cleanup_old_events_by_partition( total_partitions = len(partitions) # Both events tables partition by toYYYYMM(timestamp), so the same partition list applies; # deleting IN PARTITION on a partition a table doesn't have is a no-op. + # + # Events only, deliberately: this enforces a multi-year retention floor for a named set of + # teams, and every other personal-data table already expires sooner under its own TTL. event_tables = events_data_tables(cluster) for idx, partition in enumerate(partitions, 1): diff --git a/posthog/dags/tests/conftest.py b/posthog/dags/tests/conftest.py index 36b5ef437384..08b093737411 100644 --- a/posthog/dags/tests/conftest.py +++ b/posthog/dags/tests/conftest.py @@ -4,6 +4,8 @@ __all__ = ["django_db_setup"] from collections.abc import Iterator +from datetime import datetime +from uuid import UUID import pytest from posthog.test.base import reset_clickhouse_database @@ -11,6 +13,7 @@ from django.conf import settings +from clickhouse_driver import Client from psycopg.types.json import Jsonb from posthog.clickhouse.cluster import ClickhouseCluster, get_cluster @@ -25,6 +28,14 @@ from posthog.persons_db import persons_db_connection +def insert_flag_evaluations(rows: list[tuple[int, str, str | UUID, str | UUID, datetime]], client: Client) -> None: + """Insert rows of (team_id, distinct_id, person_id, uuid, timestamp) into flag_evaluations.""" + client.execute( + "INSERT INTO writable_flag_evaluations (team_id, distinct_id, person_id, uuid, timestamp) VALUES", + rows, + ) + + def refresh_person_from_persons_db(person) -> None: """Reload a Person instance's mutable fields from the persons DB. diff --git a/posthog/dags/tests/test_data_deletion_requests.py b/posthog/dags/tests/test_data_deletion_requests.py index d9cd64697809..870ee91f63e0 100644 --- a/posthog/dags/tests/test_data_deletion_requests.py +++ b/posthog/dags/tests/test_data_deletion_requests.py @@ -31,12 +31,14 @@ delete_person_recordings_op, execute_event_deletion, finalize_deletion_request, + get_property_removal_shards, load_deletion_request, load_person_removal_request, load_property_removal_request, process_property_removal_shard, verify_property_removal, ) +from posthog.dags.tests.conftest import insert_flag_evaluations from posthog.models.data_deletion_request import ( DataDeletionRequest, ExecutionMode, @@ -44,6 +46,7 @@ RequestType, auto_approve_pending_requests, ) +from posthog.models.flag_evaluations.sql import FLAG_EVALUATIONS_SOURCE_EVENT from posthog.test.persons import create_person TEAM_ID = 99999 @@ -91,6 +94,33 @@ def _truncate_writable_events(client: Client) -> None: client.execute("TRUNCATE TABLE IF EXISTS sharded_events") +def _insert_flag_evaluations_with_properties(rows: list[tuple], client: Client) -> None: + # Rows of (team_id, distinct_id, properties_json, uuid, timestamp, inserted_at). The event name + # is stamped here because a property-removal request narrows on it. inserted_at is set + # explicitly rather than left to its `DEFAULT timestamp` so a test can place a row before or + # after the removal marker, which is what the marker-bounded gate keys on. + client.execute( + "INSERT INTO writable_flag_evaluations " + "(team_id, distinct_id, properties, uuid, timestamp, inserted_at, event) VALUES", + [(*row, FLAG_EVALUATIONS_SOURCE_EVENT) for row in rows], + ) + + +def _flag_evaluation_person_ids(team_id: int, client: Client) -> list[str]: + # _row_exists = 1 excludes lightweight-deleted rows, which linger until a merge; without it a + # swept row still reads back and the equality assertion passes on merge timing. + result = client.execute( + "SELECT DISTINCT toString(person_id) FROM flag_evaluations " + "WHERE team_id = %(team_id)s AND _row_exists = 1 ORDER BY person_id", + {"team_id": team_id}, + ) + return [row[0] for row in result] + + +def _truncate_flag_evaluations(client: Client) -> None: + client.execute("TRUNCATE TABLE IF EXISTS sharded_flag_evaluations") + + def _truncate_adhoc_events_deletion(client: Client) -> None: client.execute(f"TRUNCATE TABLE IF EXISTS {ADHOC_EVENTS_DELETION_TABLE}") @@ -1915,6 +1945,7 @@ def test_delete_person_events_op_runs_lightweight_delete_per_shard(cluster: Clic now = datetime.now() cluster.any_host(_truncate_writable_events).result() + cluster.any_host(_truncate_flag_evaluations).result() cluster.any_host( partial( _insert_events_with_person, @@ -1924,6 +1955,16 @@ def test_delete_person_events_op_runs_lightweight_delete_per_shard(cluster: Clic ], ) ).result() + # flag_evaluations rides the same person_id-keyed sweep, so a person deletion must reach it too. + cluster.any_host( + partial( + insert_flag_evaluations, + [ + (TEAM_ID, "erase", person_uuid, str(uuid4()), now), + (TEAM_ID, "keep", other_uuid, str(uuid4()), now), + ], + ) + ).result() ctx = PersonRemovalContext( request_id=str(uuid4()), @@ -1939,6 +1980,8 @@ def test_delete_person_events_op_runs_lightweight_delete_per_shard(cluster: Clic assert cluster.any_host(partial(_count_events_for_person, TEAM_ID, person_uuid)).result() == 0 assert cluster.any_host(partial(_count_events_for_person, TEAM_ID, other_uuid)).result() == 1 + surviving = cluster.any_host(partial(_flag_evaluation_person_ids, TEAM_ID)).result() + assert surviving == [other_uuid] @pytest.mark.django_db @@ -1955,6 +1998,111 @@ def test_delete_person_events_op_noop_when_disabled(cluster: ClickhouseCluster): delete_person_events_op(build_op_context(), cluster, ctx) +@pytest.mark.django_db +def test_deferred_event_removal_queues_flag_evaluations_and_blocks_promotion(cluster: ClickhouseCluster): + # Every other deferred test names an event flag_evaluations can't hold, so stored_events skips + # the table and none of this code runs. A request naming $feature_flag_called must queue the + # table's own uuids, and must not be promoted to COMPLETED while its rows survive. + now = datetime.now() + start_time = now - timedelta(days=7) + end_time = now + timedelta(minutes=1) + request = DataDeletionRequest.objects.create( + team_id=DEFERRED_TEAM_ID, + request_type=RequestType.EVENT_REMOVAL, + events=[FLAG_EVALUATIONS_SOURCE_EVENT], + start_time=start_time, + end_time=end_time, + status=RequestStatus.QUEUED, + execution_mode=ExecutionMode.DEFERRED.value, + ) + + cluster.any_host(_truncate_adhoc_events_deletion).result() + cluster.any_host(_truncate_flag_evaluations).result() + + # A uuid that exists only in flag_evaluations: reading candidates from the events table alone + # would never queue it. + flag_uuid = uuid4() + cluster.any_host( + partial( + _insert_flag_evaluations_with_properties, + [(DEFERRED_TEAM_ID, "someone", "{}", str(flag_uuid), now, now)], + ) + ).result() + + deletion_ctx = DeletionRequestContext( + request_id=str(request.pk), + team_id=DEFERRED_TEAM_ID, + start_time=start_time, + end_time=end_time, + events=[FLAG_EVALUATIONS_SOURCE_EVENT], + execution_mode=ExecutionMode.DEFERRED.value, + ) + execute_event_deletion(build_op_context(), cluster, deletion_ctx) + + queued = cluster.any_host(partial(_adhoc_pending_uuids, DEFERRED_TEAM_ID)).result() + assert flag_uuid in queued + + # The row is still there until the drain runs, so verification must not promote the request. + from posthog.models.data_deletion_request import verify_queued_request + + outcome = verify_queued_request(request) + assert outcome.remaining == 1 + assert outcome.promoted is False + + cluster.any_host(_truncate_adhoc_events_deletion).result() + + +@pytest.mark.django_db +def test_get_property_removal_shards_refuses_when_flag_evaluations_holds_matching_rows(cluster: ClickhouseCluster): + # Property removal cannot rewrite flag_evaluations: its typed columns are MATERIALIZED, so + # ClickHouse rejects both an assignment to them and an update of properties itself. Completing + # the request anyway would report the property erased while a copy of it survived. + request = DataDeletionRequest.objects.create( + team_id=PROP_TEAM_ID, + request_type=RequestType.PROPERTY_REMOVAL, + events=[FLAG_EVALUATIONS_SOURCE_EVENT], + properties=["$ip"], + start_time=datetime.now() - timedelta(days=7), + end_time=datetime.now() + timedelta(days=1), + status=RequestStatus.APPROVED, + ) + config = DataDeletionRequestConfig(request_id=str(request.pk)) + deletion_request = load_property_removal_request(build_op_context(), config) + marker = deletion_request.inserted_at_marker + assert marker is not None + before_marker = marker - timedelta(minutes=1) + after_marker = marker + timedelta(minutes=1) + + cluster.any_host(_truncate_flag_evaluations).result() + cluster.any_host( + partial( + _insert_flag_evaluations_with_properties, + [(PROP_TEAM_ID, "someone", '{"$ip": "1.2.3.4"}', str(uuid4()), before_marker, before_marker)], + ) + ).result() + + # Match the refusal itself, not just the table name: an UNKNOWN_TABLE error would also carry + # "flag_evaluations" and would let a broken gate pass this test. + with pytest.raises(dagster.Failure, match="cannot be deleted"): + list(get_property_removal_shards(build_op_context(), cluster, deletion_request)) + + # A row ingested after the marker is outside the deletion snapshot, so the gate must ignore it + # rather than refuse forever. Without the marker bound this row wedges every retry. + cluster.any_host(_truncate_flag_evaluations).result() + cluster.any_host( + partial( + _insert_flag_evaluations_with_properties, + [(PROP_TEAM_ID, "someone", '{"$ip": "1.2.3.4"}', str(uuid4()), after_marker, after_marker)], + ) + ).result() + assert list(get_property_removal_shards(build_op_context(), cluster, deletion_request)) + + # With nothing left to strand the same request proceeds, so the gate cannot degrade into an + # unconditional block on every property removal. + cluster.any_host(_truncate_flag_evaluations).result() + assert list(get_property_removal_shards(build_op_context(), cluster, deletion_request)) + + @pytest.mark.django_db def test_delete_person_events_op_resolves_distinct_ids_to_uuids(cluster: ClickhouseCluster): # When the request was submitted by distinct_id, the events op must resolve to UUIDs diff --git a/posthog/dags/tests/test_deletes.py b/posthog/dags/tests/test_deletes.py index bebfdb16ed4f..898afb355489 100644 --- a/posthog/dags/tests/test_deletes.py +++ b/posthog/dags/tests/test_deletes.py @@ -20,10 +20,20 @@ find_partitions_to_cleanup, monthly_old_events_cleanup_job, ) +from posthog.dags.tests.conftest import insert_flag_evaluations from posthog.models.async_deletion import AsyncDeletion, DeletionType from posthog.models.person.sql import PERSON_DISTINCT_ID_OVERRIDES_TABLE +def surviving_flag_evaluations(client: Client) -> set[tuple[int, UUID, UUID]]: + # _row_exists = 1 filters out lightweight-deleted rows, which stay visible until a merge. Without + # it a swept row still reads back and the "survived the sweep" assertions pass on merge timing. + result = client.execute("SELECT team_id, person_id, uuid FROM flag_evaluations WHERE _row_exists = 1") + if not isinstance(result, list): + return set() + return {(row[0], row[1], row[2]) for row in result} + + @pytest.mark.django_db def test_full_job_person_deletes(cluster: ClickhouseCluster): timestamp = (datetime.now() + timedelta(days=31)).replace( @@ -45,6 +55,20 @@ def insert_events(client: Client) -> None: cluster.any_host(insert_events).result() + # Flag-evaluation rows for one person whose deletion lands inside the window and one whose + # deletion sits past the override watermark, mirroring the two event assertions below. The + # sweep reaches this table only because it is registered as a deletion target. + swept_person, retained_person = hour_delay + 2, hour_delay - 2 + cluster.any_host( + partial( + insert_flag_evaluations, + [ + (i, f"distinct_id_{i}", UUID(int=i), UUID(int=i), timestamp - timedelta(hours=i)) + for i in (swept_person, retained_person) + ], + ) + ).result() + def get_oldest_override_timestamp(client: Client) -> datetime: result = client.execute(f"SELECT min(_timestamp) FROM {PERSON_DISTINCT_ID_OVERRIDES_TABLE}") if not result or result[0][0] is None: @@ -119,6 +143,14 @@ def get_pending_deletes() -> list[AsyncDeletion]: f"Expected UUID {deleted_uuid} to be deleted" ) + surviving = cluster.any_host(surviving_flag_evaluations).result() + assert (swept_person, UUID(int=swept_person), UUID(int=swept_person)) not in surviving, ( + "flag_evaluations rows for a deleted person survived the sweep" + ) + assert (retained_person, UUID(int=retained_person), UUID(int=retained_person)) in surviving, ( + "the sweep deleted a flag_evaluations row it should not have" + ) + # Verify that the deletions before oldest override timestamp have been marked verified pre_override_deletions = AsyncDeletion.objects.filter(created_at__lte=oldest_override_timestamp) assert len(pre_override_deletions) == delete_count - hour_delay @@ -378,9 +410,21 @@ def get_optimized_rows(client: Client) -> int: cluster.any_host(insert_events).result() cluster.any_host(insert_adhoc_event_deletes).result() + # The queue holds (team_id, uuid) and the drain applies it to every personal-data table, so a + # flag-evaluation row sharing a queued uuid has to go with the event. This is the assumption + # the producer must honor: mirror the source event's uuid, or the two tables diverge here. + queued_uuid, unqueued_uuid = 5, delete_count + 5 + cluster.any_host( + partial( + insert_flag_evaluations, + [(i, f"distinct_id_{i}", UUID(int=i), UUID(int=i), timestamp) for i in (queued_uuid, unqueued_uuid)], + ) + ).result() + # Check preconditions initial_events = cluster.any_host(partial(get_by_team_and_uuid, "writable_events")).result() assert len(initial_events) == event_count # All events present initially + assert len(cluster.any_host(surviving_flag_evaluations).result()) == 2 pending_deletes = cluster.any_host(get_pending_deletes).result() assert pending_deletes == delete_count @@ -406,6 +450,14 @@ def get_optimized_rows(client: Client) -> int: assert all( (event[0], event[2]) in final_events.keys() for event in events if event[0] not in range(delete_count) ), f"There are non-requested deleted events that were deleted" + + surviving = cluster.any_host(surviving_flag_evaluations).result() + assert (queued_uuid, UUID(int=queued_uuid), UUID(int=queued_uuid)) not in surviving, ( + "a flag_evaluations row whose uuid was queued for deletion survived" + ) + assert (unqueued_uuid, UUID(int=unqueued_uuid), UUID(int=unqueued_uuid)) in surviving, ( + "the drain deleted a flag_evaluations row whose uuid was never queued" + ) # Verify the temporary tables were cleaned up deletes_dict = AdhocEventDeletesDictionary(source=AdhocEventDeletesTable()) assert not any(cluster.map_all_hosts(deletes_dict.exists).result().values()) diff --git a/posthog/models/data_deletion_request.py b/posthog/models/data_deletion_request.py index d37753fdf1b0..966da94408ee 100644 --- a/posthog/models/data_deletion_request.py +++ b/posthog/models/data_deletion_request.py @@ -188,6 +188,20 @@ def event_match_params(obj) -> dict: _EVENT_REMOVAL_TIME_PREDICATE = "team_id = %(team_id)s AND timestamp >= %(start_time)s AND timestamp < %(end_time)s" +def portable_event_removal_where(obj) -> tuple[str, dict]: + """The part of an event-removal predicate every deletion target can run. + + Team/timestamp bounds plus the event-name filter — columns that every table carrying event + rows declares. Excludes the compiled HogQL fragment, which resolves against the events schema + and names physical columns (``mat_*``, the property-group maps) other tables do not have. + + Because it drops a narrowing clause, this matches a superset of ``event_removal_where``. That + makes it safe for asking "would this request strand rows here?" and unsafe for deleting. + """ + parts = [_EVENT_REMOVAL_TIME_PREDICATE, event_match_sql_fragment(obj)] + return " ".join(p for p in parts if p), event_match_params(obj) + + def event_removal_where(obj, use_new_events_schema: bool = False) -> tuple[str, dict]: """Full WHERE predicate + params for event-removal queries. @@ -198,8 +212,8 @@ def event_removal_where(obj, use_new_events_schema: bool = False) -> tuple[str, proxy or the local ``sharded_events`` MergeTree. Pass ``use_new_events_schema`` when the query targets the native-JSON events tables. """ - parts = [_EVENT_REMOVAL_TIME_PREDICATE, event_match_sql_fragment(obj)] - params = event_match_params(obj) + predicate, params = portable_event_removal_where(obj) + parts = [predicate] hogql_sql, hogql_values = compile_hogql_predicate(obj, use_new_events_schema=use_new_events_schema) if hogql_sql: parts.append(f"AND ({hogql_sql})") @@ -742,18 +756,23 @@ def refresh_deletion_stats(request: "DataDeletionRequest", *, user_id: int | Non def count_remaining_matching_events(request: "DataDeletionRequest") -> int: - """Count events still matching an event-removal request's criteria in ClickHouse. + """Count rows still matching an event-removal request's criteria in ClickHouse. + + Counts across every registered read table that could hold the named events. A request is only + complete once its rows are gone from all of them. - Counts across every events read table (legacy and native-JSON) — a request is only complete - once its events are gone from all of them. + A target that cannot take the compiled HogQL fragment is counted with the portable predicate + instead, which matches a superset. That can only hold a request in QUEUED, never promote one + early. It also means a non-zero count is not proof that rows were missed: for a HogQL request + the superset can match rows the predicate itself never would. """ from posthog.clickhouse.client import sync_execute from posthog.clickhouse.client.connection import ClickHouseUser from posthog.clickhouse.query_tagging import Feature, Product, tags_context from posthog.clickhouse.workload import Workload - from posthog.models.event.deletion import events_read_tables_via_sync_execute - from posthog.models.event.sql import DISTRIBUTED_EVENTS_JSON_TABLE + from posthog.models.deletion_targets import resolve_read_targets_via_sync_execute, surviving_rows_sql + events = [] if request.delete_all_events else request.events total = 0 with tags_context( product=Product.INTERNAL, @@ -762,13 +781,15 @@ def count_remaining_matching_events(request: "DataDeletionRequest") -> int: workload=Workload.OFFLINE, query_type="data_deletion_request_verify_queued", ): - for table in events_read_tables_via_sync_execute(): - predicate, params = event_removal_where( - request, use_new_events_schema=table == DISTRIBUTED_EVENTS_JSON_TABLE - ) - # nosemgrep: clickhouse-fstring-param-audit (predicate built from internal helper, not user input) + for target in resolve_read_targets_via_sync_execute(): + if not target.may_hold_any_of(events): + continue + if target.accepts_hogql_predicate: + predicate, params = event_removal_where(request, use_new_events_schema=target.uses_new_events_schema) + else: + predicate, params = portable_event_removal_where(request) result = sync_execute( - f"SELECT count() FROM {table} WHERE {predicate} AND _row_exists = 1", + surviving_rows_sql(target.read_table, predicate), params, team_id=request.team_id, readonly=True, @@ -795,6 +816,9 @@ def discover_affected_mat_columns(properties: list[str], table_column: str) -> l ``column_materializer::::`` convention. Mirrors ``_get_affected_mat_columns`` in the deletion job so verification counts a row as dirty on the same terms the deletion does — a value left in a materialized column after its JSON key is gone still counts. + + Scoped to ``events`` deliberately, matching the deletion job; see + docs/internal/clickhouse-deletion-coverage.md. """ if not properties: return [] diff --git a/posthog/models/deletion_targets.py b/posthog/models/deletion_targets.py new file mode 100644 index 000000000000..3f3713d1f2a5 --- /dev/null +++ b/posthog/models/deletion_targets.py @@ -0,0 +1,237 @@ +"""Which physical ClickHouse tables a data deletion has to sweep. + +Erasing a person's data is not a property of the events table — it is a property of every table +that stores rows attributable to a person. This module is the one list of those tables, so a table +that starts carrying person data gets registered here rather than found by an auditor months later. + +Every target is swept with predicates built only from columns all of them declare: ``team_id``, +``person_id``, ``distinct_id``, ``timestamp``, ``uuid``, ``event``. + +The reasoning behind each registration, exclusion and known gap is in +docs/internal/clickhouse-deletion-coverage.md. +""" + +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum +from functools import partial + +from django.conf import settings + +from clickhouse_driver import Client + +from posthog.clickhouse.client import sync_execute +from posthog.clickhouse.client.connection import NodeRole +from posthog.clickhouse.cluster import ClickhouseCluster, Query +from posthog.models.event.sql import ( + DISTRIBUTED_EVENTS_JSON_TABLE, + EVENTS_DATA_TABLE, + EVENTS_JSON_DATA_TABLE, + SHARDED_EVENTS_RECENT_DATA_TABLE, +) +from posthog.models.flag_evaluations.sql import ( + FLAG_EVALUATIONS_DATA_TABLE, + FLAG_EVALUATIONS_SOURCE_EVENT, + FLAG_EVALUATIONS_TABLE, +) + + +class UnsweepableRowsError(Exception): + """A deletion would report success while rows it cannot reach survive on another table.""" + + +class HogQLSchema(Enum): + """Which schema a compiled HogQL predicate must target to run against a table.""" + + LEGACY = "legacy" + NATIVE_JSON = "native_json" + + +@dataclass(frozen=True, kw_only=True) +class DeletionTarget: + """One physical table a deletion sweep runs against. + + ``data_table`` is the local per-shard storage table a mutation is issued on; ``read_table`` is + the Distributed proxy a verification count reads through. + + The capability fields exist so a caller needing more than the schema-agnostic predicates fails + loudly on a table that cannot take it, rather than skipping it silently. + """ + + data_table: str + read_table: str + # Guarded on system.tables: the table sits behind a migration that may not have run everywhere. + optional: bool = False + # The schema a HogQL predicate compiles against here, None where no HogQL table definition + # exists. A compiled fragment names physical columns (mat_*, the property-group maps, or JSON + # subcolumns), so it only runs against the schema it was compiled for. + hogql_schema: HogQLSchema | None = None + # properties/person_properties can be rewritten in place. True only where the property columns + # are DEFAULT-kind, as materialize() mints them, and so can be reset by ALTER UPDATE. + accepts_property_rewrite: bool = False + # Read uuids from this table when queueing a deferred deletion. False where the rows duplicate + # another target's uuids, which would queue each one twice. + queue_uuid_candidates: bool = True + # The event names this table can hold, None meaning unconstrained. Lets a request naming other + # events skip this table without querying it. + stored_events: frozenset[str] | None = None + + @property + def accepts_hogql_predicate(self) -> bool: + return self.hogql_schema is not None + + @property + def uses_new_events_schema(self) -> bool: + return self.hogql_schema is HogQLSchema.NATIVE_JSON + + def may_hold_any_of(self, events: Sequence[str]) -> bool: + """Whether this table could hold rows for any of ``events``, empty meaning every event. + + A request naming only events a table never stores can skip it, which matters because + counting to prove zero is not free: no target indexes ``event``, so the query reads the + team's rows across whichever partitions the request's time range covers. + """ + if not events or self.stored_events is None: + return True + return not self.stored_events.isdisjoint(events) + + +EVENTS = DeletionTarget( + data_table=EVENTS_DATA_TABLE(), + read_table="events", + hogql_schema=HogQLSchema.LEGACY, + accepts_property_rewrite=True, +) + +EVENTS_JSON = DeletionTarget( + data_table=EVENTS_JSON_DATA_TABLE, + read_table=DISTRIBUTED_EVENTS_JSON_TABLE, + optional=True, + hogql_schema=HogQLSchema.NATIVE_JSON, + accepts_property_rewrite=True, + # Dual-written from the same events, so its uuids are the legacy table's. + queue_uuid_candidates=False, +) + +# Flag-evaluation telemetry carries the same person and group payload as events, so person, team +# and queued-uuid sweeps must reach it. It takes neither of the richer sweeps; both exclusions are +# explained in docs/internal/clickhouse-deletion-coverage.md. Its producer populates person_id +# exactly as the events pipeline does, so a person sweep matches on person_id like every other +# events-shaped table. +FLAG_EVALUATIONS = DeletionTarget( + data_table=FLAG_EVALUATIONS_DATA_TABLE, + read_table=FLAG_EVALUATIONS_TABLE, + optional=True, + stored_events=frozenset({FLAG_EVALUATIONS_SOURCE_EVENT}), +) + +EVENTS_TARGETS: tuple[DeletionTarget, ...] = (EVENTS, EVENTS_JSON) +PERSONAL_DATA_TARGETS: tuple[DeletionTarget, ...] = (*EVENTS_TARGETS, FLAG_EVALUATIONS) + +# Storage tables that carry person properties and are reclaimed by their TTL alone. Each entry is a +# decision that erasure may lag by the retention window, not an oversight. +# +# sharded_events_recent is a transient mirror of the last few days of events, on a 7-day TTL keyed +# on inserted_at. Seven days is a short enough window to accept as the erasure bound, and a sweep +# would race the TTL for little benefit. +TTL_ONLY_TABLES: frozenset[str] = frozenset({SHARDED_EVENTS_RECENT_DATA_TABLE()}) + + +_TABLE_EXISTS_SQL = "SELECT count() FROM system.tables WHERE database = %(database)s AND name = %(name)s" + + +def _table_exists_params(table: str) -> dict[str, str]: + return {"database": settings.CLICKHOUSE_DATABASE, "name": table} + + +def _table_exists(client: Client, table: str) -> bool: + [[count]] = client.execute(_TABLE_EXISTS_SQL, _table_exists_params(table)) + return bool(count) + + +def _table_exists_via_sync_execute(table: str) -> bool: + result = sync_execute(_TABLE_EXISTS_SQL, _table_exists_params(table)) + return bool(result and result[0][0]) + + +def is_present(cluster: ClickhouseCluster, target: DeletionTarget) -> bool: + """Whether any data node carries this target's storage table. + + ``any`` rather than ``all``: on a partially-migrated cluster the mutation fails loudly on the + hosts missing the table, which is preferable to silently skipping a deletion. + """ + if not target.optional: + return True + results = cluster.map_hosts_by_role(partial(_table_exists, table=target.data_table), NodeRole.DATA).result() + return any(results.values()) + + +def resolve_targets( + cluster: ClickhouseCluster, targets: Sequence[DeletionTarget] = PERSONAL_DATA_TARGETS +) -> list[DeletionTarget]: + """The subset of ``targets`` that actually exists on this cluster.""" + return [target for target in targets if is_present(cluster, target)] + + +def personal_data_tables(cluster: ClickhouseCluster) -> list[str]: + """Every physical table a person, team, or queued-uuid deletion must sweep.""" + return [target.data_table for target in resolve_targets(cluster)] + + +def resolve_data_targets_via_sync_execute(targets: Sequence[DeletionTarget]) -> list[DeletionTarget]: + """Targets whose storage table exists, for callers talking to ClickHouse via ``sync_execute``.""" + return [target for target in targets if not target.optional or _table_exists_via_sync_execute(target.data_table)] + + +def resolve_read_targets_via_sync_execute( + targets: Sequence[DeletionTarget] = PERSONAL_DATA_TARGETS, +) -> list[DeletionTarget]: + """Targets whose Distributed read table exists, for callers verifying a deletion completed.""" + return [target for target in targets if not target.optional or _table_exists_via_sync_execute(target.read_table)] + + +def surviving_rows_sql(read_table: str, predicate: str) -> str: + """Count rows still matching ``predicate``, excluding ones a lightweight delete already hid. + + ``_row_exists = 1`` is what makes this a survivor count rather than a row count; without it + every deleted row still reads as present until its parts merge. + """ + # nosemgrep: clickhouse-fstring-param-audit (predicate built from internal helpers, not user input) + return f"SELECT count() FROM {read_table} WHERE {predicate} AND _row_exists = 1" + + +def assert_no_unsweepable_rows( + cluster: ClickhouseCluster, + targets: Sequence[DeletionTarget], + predicate: str, + params: dict, + *, + events: Sequence[str], + reason: str, +) -> None: + """Fail when a target this request cannot sweep actually holds rows matching it. + + Pass targets already narrowed by ``resolve_targets``; this does not re-check presence. + + ``predicate`` must be the portable part of the criteria — no HogQL fragment, no + materialized-column arms — so the count is a superset of what the deletion would have removed. + ``events`` is the request's event filter, empty meaning every event. + + Deliberately gated on rows existing rather than on the request's shape: an unconditional + refusal would block every request of that shape from the day it lands, including the ones with + nothing to strand. + """ + for target in targets: + if not target.may_hold_any_of(events): + continue + + # Read through the Distributed proxy on a data node — these tables exist on no other role. + # Bounded like the job's other scans: no target indexes `event`, so this reads the team's + # rows across the request's partitions once the table is no longer empty. + query = Query(surviving_rows_sql(target.read_table, predicate), params, settings={"max_execution_time": "1800"}) + rows = cluster.any_host_by_role(query, NodeRole.DATA).result() + count = int(rows[0][0]) if rows else 0 + if count: + raise UnsweepableRowsError( + f"{count} row(s) in {target.read_table} match this request but cannot be deleted: {reason}" + ) diff --git a/posthog/models/event/deletion.py b/posthog/models/event/deletion.py index 52decdd99805..ee16a85855e6 100644 --- a/posthog/models/event/deletion.py +++ b/posthog/models/event/deletion.py @@ -6,65 +6,38 @@ diverge and deleted data survives in whichever table serves queries. The JSON tables are only targeted where they exist, so environments that have not run the ``0288_events_json_schema`` migration (and legacy-schema test databases) keep working unchanged. -""" - -from functools import partial - -from django.conf import settings -from clickhouse_driver import Client +These are the events tables specifically. A deletion that must reach every table carrying person +data — person removal, team deletion, the queued-uuid drain — wants +``posthog.models.deletion_targets.personal_data_tables`` instead. +""" -from posthog.clickhouse.client import sync_execute -from posthog.clickhouse.client.connection import NodeRole from posthog.clickhouse.cluster import ClickhouseCluster -from posthog.models.event.sql import DISTRIBUTED_EVENTS_JSON_TABLE, EVENTS_DATA_TABLE, EVENTS_JSON_DATA_TABLE - - -def _table_exists(client: Client, table: str) -> bool: - [[count]] = client.execute( - "SELECT count() FROM system.tables WHERE database = %(database)s AND name = %(name)s", - {"database": settings.CLICKHOUSE_DATABASE, "name": table}, - ) - return bool(count) - - -def _table_exists_via_sync_execute(table: str) -> bool: - result = sync_execute( - "SELECT count() FROM system.tables WHERE database = %(database)s AND name = %(name)s", - {"database": settings.CLICKHOUSE_DATABASE, "name": table}, - ) - return bool(result and result[0][0]) +from posthog.models.deletion_targets import ( + EVENTS_JSON, + EVENTS_TARGETS, + is_present, + resolve_data_targets_via_sync_execute, + resolve_read_targets_via_sync_execute, + resolve_targets, +) def cluster_has_events_json_table(cluster: ClickhouseCluster) -> bool: - """Whether any data node carries the native-JSON events data table. - - ``any`` rather than ``all``: on a partially-migrated cluster the follow-up mutation fails - loudly on the hosts missing the table, which is preferable to silently skipping a deletion. - """ - results = cluster.map_hosts_by_role(partial(_table_exists, table=EVENTS_JSON_DATA_TABLE), NodeRole.DATA).result() - return any(results.values()) + """Whether any data node carries the native-JSON events data table.""" + return is_present(cluster, EVENTS_JSON) def events_data_tables(cluster: ClickhouseCluster) -> list[str]: """The physical events data tables that deletions/mutations must target, for Dagster jobs.""" - tables = [EVENTS_DATA_TABLE()] - if cluster_has_events_json_table(cluster): - tables.append(EVENTS_JSON_DATA_TABLE) - return tables + return [target.data_table for target in resolve_targets(cluster, EVENTS_TARGETS)] def events_data_tables_via_sync_execute() -> list[str]: """Like ``events_data_tables``, for callers that talk to ClickHouse through ``sync_execute``.""" - tables = [EVENTS_DATA_TABLE()] - if _table_exists_via_sync_execute(EVENTS_JSON_DATA_TABLE): - tables.append(EVENTS_JSON_DATA_TABLE) - return tables + return [target.data_table for target in resolve_data_targets_via_sync_execute(EVENTS_TARGETS)] def events_read_tables_via_sync_execute() -> list[str]: """The distributed events read tables to check when verifying that a deletion completed.""" - tables = ["events"] - if _table_exists_via_sync_execute(DISTRIBUTED_EVENTS_JSON_TABLE): - tables.append(DISTRIBUTED_EVENTS_JSON_TABLE) - return tables + return [target.read_table for target in resolve_read_targets_via_sync_execute(EVENTS_TARGETS)] diff --git a/posthog/models/flag_evaluations/sql.py b/posthog/models/flag_evaluations/sql.py index 7d35dcbc0d45..2ec2ccb3582e 100644 --- a/posthog/models/flag_evaluations/sql.py +++ b/posthog/models/flag_evaluations/sql.py @@ -33,6 +33,11 @@ FLAG_EVALUATIONS_TTL_DAYS = 90 +# The only event this table stores. posthog/models/deletion_targets.py uses it to skip this table +# for deletion requests that name other events, so a producer writing a second event name here has +# to update the target's stored_events. +FLAG_EVALUATIONS_SOURCE_EVENT = "$feature_flag_called" + # Sharding by a distinct_id hash rather than anything team-based: a handful of # teams carry most of the volume, and sharding by team would hotspot single # shards. sipHash64 matches the events table's Distributed sharding key, so a @@ -188,6 +193,26 @@ def FLAG_EVALUATIONS_DATA_TABLE_ENGINE() -> MergeTreeEngine: ) +def DROP_FLAG_EVALUATIONS_PROXY_TABLES_SQL() -> list[str]: + """Drop the two Distributed fronts so a reset recreates them alongside the storage table. + + Recreating the shard while leaving these on an older column list makes inserts silently drop + the new column and reads not see it, which fails as a puzzle rather than as a schema error. + """ + return [ + f"DROP TABLE IF EXISTS {FLAG_EVALUATIONS_TABLE}", + f"DROP TABLE IF EXISTS {FLAG_EVALUATIONS_WRITABLE_TABLE}", + ] + + +def DROP_FLAG_EVALUATIONS_TABLE_SQL() -> str: + # reset_clickhouse_database drops rather than truncates because MutationRunner skips enqueueing + # a mutation whose command text already exists on the table, so mutation history left behind by + # one test turns a later test's identical delete into a no-op. SYNC so the replica's ZooKeeper + # metadata is gone before the table is recreated. + return f"DROP TABLE IF EXISTS {FLAG_EVALUATIONS_DATA_TABLE} SYNC" + + def _distributed_table_sql(table_name: str, *, materialized_columns: str = "") -> str: return f""" CREATE TABLE IF NOT EXISTS {table_name} diff --git a/posthog/test/base.py b/posthog/test/base.py index 27a234eae98c..dce8a3f6956f 100644 --- a/posthog/test/base.py +++ b/posthog/test/base.py @@ -126,6 +126,13 @@ EXCHANGE_RATE_DICTIONARY_SQL, EXCHANGE_RATE_TABLE_SQL, ) +from posthog.models.flag_evaluations.sql import ( + DISTRIBUTED_FLAG_EVALUATIONS_TABLE_SQL, + DROP_FLAG_EVALUATIONS_PROXY_TABLES_SQL, + DROP_FLAG_EVALUATIONS_TABLE_SQL, + FLAG_EVALUATIONS_TABLE_SQL, + WRITABLE_FLAG_EVALUATIONS_TABLE_SQL, +) from posthog.models.group.sql import TRUNCATE_GROUPS_TABLE_SQL from posthog.models.instance_setting import get_instance_setting from posthog.models.organization import OrganizationMembership @@ -2057,6 +2064,8 @@ def reset_clickhouse_database() -> None: TRUNCATE_PLUGIN_LOG_ENTRIES_TABLE_SQL, TRUNCATE_CUSTOM_METRICS_COUNTER_EVENTS_TABLE, TRUNCATE_AI_EVENTS_TABLE_SQL(), + DROP_FLAG_EVALUATIONS_TABLE_SQL(), + *DROP_FLAG_EVALUATIONS_PROXY_TABLES_SQL(), ] ) run_clickhouse_statement_in_parallel( @@ -2064,6 +2073,9 @@ def reset_clickhouse_database() -> None: CHANNEL_DEFINITION_TABLE_SQL(), EXCHANGE_RATE_TABLE_SQL(), *clickhouse_events_data_table_sqls(), + FLAG_EVALUATIONS_TABLE_SQL(), + WRITABLE_FLAG_EVALUATIONS_TABLE_SQL(), + DISTRIBUTED_FLAG_EVALUATIONS_TABLE_SQL(), PERSONS_TABLE_SQL(), PROPERTY_DEFINITIONS_TABLE_SQL(), RAW_SESSIONS_TABLE_SQL(), From 700436e96ce17cfc429f5f1615b62accca1c1976 Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Thu, 13 Aug 2026 10:35:28 -0700 Subject: [PATCH 2/2] chore(flags): record DEFAULT-switch plan and restore sweep comment Generated-By: PostHog Desktop Task-Id: b6e91f38-dd91-4c39-bfce-322d583c221e --- docs/internal/clickhouse-deletion-coverage.md | 26 +++++++++++-------- posthog/dags/deletes.py | 5 ++-- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/docs/internal/clickhouse-deletion-coverage.md b/docs/internal/clickhouse-deletion-coverage.md index 624efcdfa981..5d255d8f25ab 100644 --- a/docs/internal/clickhouse-deletion-coverage.md +++ b/docs/internal/clickhouse-deletion-coverage.md @@ -53,26 +53,30 @@ That works because `materialize()` creates columns as `DEFAULT `, which is - Updating `properties` is rejected too, because `flag_key` is materialized from it and sits in the sort key: `Updated column 'properties' affects MATERIALIZED column 'flag_key', which is a key column`. - `CREATE TABLE tmp AS sharded_flag_evaluations ENGINE = MergeTree()` inherits the sort key and the column kinds, so the staging table hits the same rejection. -Rewriting these rows therefore needs a different algorithm: -`INSERT … SELECT` the cleaned rows and let the shard recompute the typed columns on write, then lightweight-delete the originals. -Until that exists, `get_property_removal_shards` refuses to start when the table holds rows matching the request, so a request cannot complete while data it named survives. +The planned fix is two follow-ups: + +1. Recreate the table with the typed columns as `DEFAULT `, the kind `materialize()` mints on events. + Recreating is only free while the table is empty, so this must land before the producer ships. +2. Point the events rewrite machinery (column discovery, staging rewrite, shard walk) at the table; today all of it is scoped to `events`. + +One open question for the schema follow-up: `flag_key` sits in the sort key, and ClickHouse never accepts `UPDATE` on a key column, whatever its kind. +Whether `UPDATE properties` is accepted once a `DEFAULT` key column depends on it needs measuring. +If it is still rejected, the fallback is the heavier rewrite: `INSERT … SELECT` the cleaned rows, let the shard recompute the typed columns on write, then lightweight-delete the originals. + +Until the fix exists, `get_property_removal_shards` refuses to start when the table holds rows matching the request, so a request cannot complete while data it named survives. The check costs nothing while the table is empty. This is also why `flag_evaluations` is deliberately absent from `MATERIALIZATION_VALID_TABLES`. Adding it would let `materialize()` mint `DEFAULT`-kind columns on a table whose property-removal path cannot reset them. -#### Decide the escape hatch before a producer lands +#### If a request arrives before the fix lands Today the refusal costs nothing, because the table is empty. Once it holds rows, a property removal with `delete_all_events` refuses whenever a single flag-evaluation row carries the named property, and the operator has no way through: `delete_all_events` and `events` are mutually exclusive on the model, so the request cannot be narrowed to exclude `$feature_flag_called`, and the admin Retry button replays the same failure. -The only exits are waiting out the TTL or shipping the re-insert algorithm. The table partitions by month with `ttl_only_drop_parts = 1`, so a part drops only once its newest row expires: the real wait is up to about 120 days, not the 90-day TTL. `posthog/models/flag_evaluations/sql.py` says the same thing next to the partition clause. - -That is the right default (refusing beats silently under-deleting), but it needs a deliberate answer before real traffic hits it. The options, in rough order of cost: - -1. Ship the re-insert rewrite, which removes the refusal entirely. -2. Let a request exclude event names, so an operator can scope around the table. -3. Record an explicit, audited acknowledgement on the request so an operator can accept the residue rather than being stuck. +The only exits are waiting out the TTL or shipping the fix above. The table partitions by month with `ttl_only_drop_parts = 1`, so a part drops only once its newest row expires: the real wait is up to about 120 days, not the 90-day TTL. `posthog/models/flag_evaluations/sql.py` says the same thing next to the partition clause. +Refusing beats silently under-deleting, so the gate is the right default. +If the fix has not landed by the time real traffic hits, the cheaper stopgaps are letting a request exclude event names so an operator can scope around the table, or recording an explicit, audited acknowledgement on the request so an operator can accept the residue rather than being stuck. Doing nothing means the first affected GDPR request becomes an escalation. ### Event removal with a HogQL predicate does not reach `flag_evaluations` diff --git a/posthog/dags/deletes.py b/posthog/dags/deletes.py index b07903a94927..58b4da4b1abe 100644 --- a/posthog/dags/deletes.py +++ b/posthog/dags/deletes.py @@ -602,8 +602,9 @@ def count_pending_adhoc_deletes(client: Client) -> int: } ) - # The predicate below reads only team_id, person_id, timestamp and uuid — columns every - # registered target declares — so it applies unchanged to all of them. + # Every registered target must get this delete, or rows survive on the table that got skipped. + # The predicate below reads only team_id, person_id, timestamp and uuid, which every target + # declares, so it applies unchanged to all of them. delete_mutation_runners = [ LightweightDeleteMutationRunner( table=table,