feat: Add person distinct ID overrides squash job (as dagster job) #27710

tkaemming · 2025-01-20T23:49:40Z

Problem

Whenever a person merge occurs, we write the latest person_id that should be associated with the affected distinct_id(s) to the person_distinct_id_overrides table, and use these records to "override" the existing value in the events (or sharded_events) table's person_id column. Another helpful way to think of this overrides table is that it represents a queue of pending person_id updates to that haven't yet been applied to the events table.

The number of overrides within this table can accumulate over time to a degree where read performance can be negatively impacted due to the size of the overrides join used to update stale values at query time. This can also lead to a situation where all of a team's queries that use person_id can run into MemoryLimitExceeded errors when using an overrides-based PoE mode if the number of overrides for a team grows large enough.

Changes

This automates the manual squash process as a Dagster job.

Briefly,

The subset of overrides to be squashed are loaded into a new replicated MergeTree table that is created on all hosts that will require it for executing mutations. Using a purpose-specific replicated table for this task ensures that the overrides state can be synchronized across all hosts without needing to pause replication or use other mechanisms to ensure all hosts are using the same data as a basis for running mutations.
This table is then loaded into a dictionary for use in mutations. Since the dictionary must be built on each host independently, we also checksum the table contents to ensure that all hosts are synchronized before executing the mutations for an additional layer of safety.
This dictionary is used in mutations that update the person_id column for all events that with a distinct_id in the overrides dictionary, and afterwards the same overrides dictionary is used to identify the rows that can be safely deleted from the main overrides table as they are no longer needed to "override" anything (the override rows now match the values in the base table.)
Once all mutations have completed, the dictionary and snapshot table are dropped on all hosts where they were created.

A few implementation notes:

The table and dictionary are run-specific and include the run ID as part of the table (or dictionary) name. This reduces the possibility for interference if two jobs are rum simultaneously, and makes it easier to find mutations in the mutations system table as we can search by the table name associated with the specific run.
Mutations apply to the full dataset and cannot be limited by partition or team ID. This allows deleting from the override table by row version alone without other considerations.
Mutations are targeted directly at relevant hosts, instead of ON CLUSTER with distributed DDL. This allows partial failure modes to be cleanly handled on retries without inadvertently scheduling redundant mutations. Mutations that are killed can be restarted, however, without needing to restart the whole job from the beginning.
Clickhouse client options can be provided as part of the job run configuration, providing more control over query execution without requiring code or configuration deploys.

I also tried to keep a lot of the design and implementation relatively orchestrator-agnostic in case we need to replatform this again.

Does this work well for both Cloud and self-hosted?

Yes.

How did you test this code?

Added test (we also ran this manually to verify the queries and general idea work.)

… of retries

… on its own

dags/person_overrides.py

tkaemming · 2025-01-21T00:08:52Z

dags/person_overrides.py

+
+    def wait(self, client: Client) -> None:
+        while not self.is_done(client):
+            time.sleep(15.0)


Arbitrarily chosen number, could be more, could be less.

dags/person_overrides.py

tkaemming · 2025-01-21T00:13:13Z

ee/clickhouse/materialized_columns/columns.py

-def get_cluster() -> ClickhouseCluster:
-    extra_hosts = []
-    for host_config in map(copy, CLICKHOUSE_PER_TEAM_SETTINGS.values()):
-        extra_hosts.append(ConnectionInfo(host_config.pop("host")))
-        assert len(host_config) == 0, f"unexpected values: {host_config!r}"
-    return ClickhouseCluster(default_client(), extra_hosts=extra_hosts)


Moved this to the non-ee package.

tkaemming · 2025-01-21T00:15:34Z

posthog/clickhouse/client/connection.py

@@ -156,8 +157,7 @@ def default_client(host=settings.CLICKHOUSE_HOST):
    )


-@cache
-def make_ch_pool(**overrides) -> ChPool:
+def _make_ch_pool(*, client_settings: Mapping[str, str] | None = None, **overrides) -> ChPool:


This rename was necessary to allow bypassing the pool cache: the client settings map isn't hashable and therefore not usable in a cache key.

tkaemming · 2025-01-21T00:20:52Z

posthog/conftest.py

@@ -7,7 +7,7 @@
 from posthog.test.base import PostHogTestCase, run_clickhouse_statement_in_parallel


-def create_clickhouse_tables(num_tables: int):


This change (and the other test changes to/around create_clickhouse_tables) wasn't actually necessary: I thought I as going to need to do more invasive fixture surgery than just the dag_tests/conftest.py edit and found the coupling to django_db_setup confusing so I simplified the logic here before realizing I wouldn't need to call this function directly.

Daesgar

Looking good!

I tried to follow along with the current manual squash process. Love your extremely clean code.

Left some questions to understand the process better, but nothing blocking!

dags/person_overrides.py

* master: (103 commits) feat(postgres-estimated-rows): pg Estimated Rows on Data Warehouse Sync (#27634) fix: revert darkmode class toggle, updated content on fills (#27783) chore: upgrade posthog-js (#27790) chore(editor-3001): add back join actions (#27740) feat: Add person distinct ID overrides squash job (as dagster job) (#27710) fix(created-by-sources): Adding `created_by` to sources (#27751) Revert "feat(data-warehouse): V2 pipeline release " (#27791) fix: typo for feature flags (#27786) fix(defer-unmounting): Defer unmounting of react elements (#27742) feat(data-warehouse): V2 pipeline release (#27732) fix(data-warehouse): Ensure dates are actual datetime formats (#27777) fix: enable hot reload for the products dir (#27746) fix: assignee selector when null (#27737) chore: clarify rrweb imports (#27776) chore(deps): Update posthog-js to 1.207.3 (#27779) feat(retention): filters on start/return event (#27770) fix(experiments): only show supported math functions (#27589) feat(web-analytics): Set unique conversions graph when adding conversions goal (#27774) chore: color design system part 1: banner and accents (#27756) chore(experiments): Add tests for funnel attribution options (#27752) ...

…27710)

tkaemming added 30 commits January 15, 2025 15:42

wip

b75e105

more queries

00b4c82

mutation queries

75c359e

lots of stuff missing but runs happy path

4568d19

use run id for initial snapshot table creation

8701ab4

require the timestamp to be provided as config

16e3c1a

split create and populate steps

22a0d32

much refactoring

0cd5e70

clumsy mutation waiter

5f36948

more clumsy

418e96a

syntax

a055f0b

easy enough

1870834

split up more things that should be able to fail/retry independently

907f772

clarify naming

3ac4798

fold some excessively long lines

004573a

take a pass through to try and determine what needs to happen in case…

0111a18

… of retries

clean up how mutations are waited

25e7e29

not sure this is the best logging approach but it'll do for now

0c731b8

timestamp can actually just be a format that is parseable by clickhouse

4fdecfd

clean up dictionary loading

6ffa346

make it safe to run enqueue operations multiple times

2f9b3f5

play it safe in populate

748fc56

formatting

acb0ace

cluster as resource

4eca7e0

shards as config

35a866e

configurable max execution time for dictionary creation

bfea2c5

support passing settings through config

3672289

very permissive default settings

232a3a4

docstrings

6bdd544

tidy

a6b5d27

tkaemming and others added 11 commits January 17, 2025 17:01

make it so that clickhouse database management is a bit easier to use…

328f8ad

… on its own

"a test"

a8dd161

test cleanup actions

15389b8

Merge branch 'master' into squash-dag

ed86838

fixtures

e2dd839

add test data

88b4539

actually assert some things

9df49f4

expand test

2181485

more tidy

cb86cb2

move get_cluster to a better home

55bfa23

Update query snapshots

1907cb9

tkaemming commented Jan 21, 2025

View reviewed changes

tkaemming added 2 commits January 20, 2025 16:33

be less clever with typing

0817855

no Mapping.__or__ support so just do the other thing

b18ccf7

tkaemming requested a review from a team January 21, 2025 00:46

tkaemming marked this pull request as ready for review January 21, 2025 00:46

Daesgar approved these changes Jan 21, 2025

View reviewed changes

tkaemming added 2 commits January 21, 2025 16:00

consolidate mostly duplicate _find_existing_mutation definitions

cce62ff

Merge branch 'master' into squash-dag

e1ae63f

tkaemming merged commit db3e85d into master Jan 22, 2025
92 checks passed

tkaemming deleted the squash-dag branch January 22, 2025 16:59

timgl pushed a commit that referenced this pull request Jan 28, 2025

feat: Add person distinct ID overrides squash job (as dagster job) (#…

b26846b

…27710)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Add person distinct ID overrides squash job (as dagster job) #27710

feat: Add person distinct ID overrides squash job (as dagster job) #27710

tkaemming commented Jan 20, 2025 •

edited

Loading

tkaemming Jan 21, 2025

tkaemming Jan 21, 2025

tkaemming Jan 21, 2025

tkaemming Jan 21, 2025

Daesgar left a comment

		@@ -7,7 +7,7 @@
		from posthog.test.base import PostHogTestCase, run_clickhouse_statement_in_parallel


		def create_clickhouse_tables(num_tables: int):

feat: Add person distinct ID overrides squash job (as dagster job) #27710

feat: Add person distinct ID overrides squash job (as dagster job) #27710

Conversation

tkaemming commented Jan 20, 2025 • edited Loading

Problem

Changes

Does this work well for both Cloud and self-hosted?

How did you test this code?

tkaemming Jan 21, 2025

Choose a reason for hiding this comment

tkaemming Jan 21, 2025

Choose a reason for hiding this comment

tkaemming Jan 21, 2025

Choose a reason for hiding this comment

tkaemming Jan 21, 2025

Choose a reason for hiding this comment

Daesgar left a comment

Choose a reason for hiding this comment

tkaemming commented Jan 20, 2025 •

edited

Loading