refactor(temporal/squash): Support flat person override table in squash workflow #19347

tkaemming · 2023-12-15T04:07:32Z

Problem

Building on top of #19220, this adds support for the "flat" person overrides model/table in the squash workflow.

Whether the existing path or flat path is used can be controlled with the postgres_person_overrides_manager workflow input parameter key. Values can either be mappings (existing path; the default) or flat (new table.)

This can be merged prior to the flat overrides write path being turned on, as it supports both the old and new paths. Once the migration occurs to use that new write path, DEFAULT_POSTGRES_PERSON_OVERRIDES_MANAGER should be updated to use flat overrides as default: otherwise, it'd try to delete from the wrong table.

How did you test this code?

Parameterized the Postgres person overrides fixtures in squash workflow tests to cover both the existing mapping implementation and the flat implementation.

tkaemming · 2023-12-18T21:35:28Z

posthog/temporal/batch_exports/squash_person_overrides.py

-SELECT_ID_FROM_OVERRIDE_UUID = """
-SELECT
-    id
-FROM
-    posthog_personoverridemapping
-WHERE
-    team_id = %(team_id)s
-    AND uuid = %(uuid)s;
-"""
-
-DELETE_FROM_PERSON_OVERRIDES = """
-DELETE FROM
-    posthog_personoverride
-WHERE
-    team_id = %(team_id)s
-    AND old_person_id = %(old_person_id)s
-    AND override_person_id = %(override_person_id)s
-    AND version = %(latest_version)s
-RETURNING
-    old_person_id;
-"""
-
-DELETE_FROM_PERSON_OVERRIDE_MAPPINGS = """
-DELETE FROM
-    posthog_personoverridemapping
-WHERE
-    id = %(id)s;
-"""


Moved these to their call site because they are now only used conditionally based on which override manager is used. (I also find them a little easier to read that way, too.)

tkaemming · 2023-12-18T21:38:02Z

posthog/temporal/batch_exports/squash_person_overrides.py

+    def delete(self, person_override: SerializablePersonOverrideToDelete, dry_run: bool = False) -> None:
+        with self.connection.cursor() as cursor:
+            SELECT_ID_FROM_OVERRIDE_UUID = """
+                SELECT
+                    id
+                FROM
+                    posthog_personoverridemapping
+                WHERE
+                    team_id = %(team_id)s
+                    AND uuid = %(uuid)s;
+            """
+
+            cursor.execute(
+                SELECT_ID_FROM_OVERRIDE_UUID,
+                {
+                    "team_id": person_override.team_id,
+                    "uuid": person_override.old_person_id,
+                },
+            )
+            row = cursor.fetchone()
+            if not row:
+                return
+
+            old_person_id = row[0]
+
+            cursor.execute(
+                SELECT_ID_FROM_OVERRIDE_UUID,
+                {
+                    "team_id": person_override.team_id,
+                    "uuid": person_override.override_person_id,
+                },
+            )
+            row = cursor.fetchone()
+            if not row:
+                return
+
+            override_person_id = row[0]
+
+            DELETE_FROM_PERSON_OVERRIDES = """
+                DELETE FROM
+                    posthog_personoverride
+                WHERE
+                    team_id = %(team_id)s
+                    AND old_person_id = %(old_person_id)s
+                    AND override_person_id = %(override_person_id)s
+                    AND version = %(latest_version)s
+                RETURNING
+                    old_person_id;
+            """
+
+            parameters = {
+                "team_id": person_override.team_id,
+                "old_person_id": old_person_id,
+                "override_person_id": override_person_id,
+                "latest_version": person_override.latest_version,
+            }
+
+            if dry_run is True:
+                activity.logger.info("This is a DRY RUN so nothing will be deleted.")
+                activity.logger.info(
+                    "Would have run query: %s with parameters %s",
+                    DELETE_FROM_PERSON_OVERRIDES,
+                    parameters,
+                )
+                return
+
+            cursor.execute(DELETE_FROM_PERSON_OVERRIDES, parameters)
+            row = cursor.fetchone()
+            if not row:
+                # There is no existing mapping for this (old_person_id, override_person_id) pair.
+                # It could be that a newer one was added (with a later version).
+                return
+
+            deleted_id = row[0]
+
+            DELETE_FROM_PERSON_OVERRIDE_MAPPINGS = """
+                DELETE FROM
+                    posthog_personoverridemapping
+                WHERE
+                    id = %(deleted_id)s;
+            """
+
+            cursor.execute(
+                DELETE_FROM_PERSON_OVERRIDE_MAPPINGS,
+                {
+                    "deleted_id": deleted_id,
+                },
+            )


Kind of a similar story to #19220 (comment): the rest of this class is mostly here to support easier substitution during tests — this is the only method that gets used on production paths.

This particular block was just copied in from the original call site.

There's also a lot of duplication here with that (JS/TS) implementation as well. Unfortunate consequence of cross-runtime logic, I guess.

tkaemming · 2023-12-18T21:40:21Z

posthog/temporal/batch_exports/squash_person_overrides.py

+    override_person_id: UUID
+
+
+class PostgresPersonOverridesManager:


Not married to this name (kind of actively dislike it) — not sure if something like PostgresPersonOverrideStrategy would make sense? Or something else completely?

No strong opinion here. PostgresPersonOverrideStrategy is good too. 🤷‍♂️

tkaemming · 2023-12-18T21:41:47Z

posthog/temporal/batch_exports/squash_person_overrides.py

+POSTGRES_PERSON_OVERRIDES_MANAGERS = {
+    "mappings": PostgresPersonOverridesManager,
+    "flat": FlatPostgresPersonOverridesManager,
+}


These technically could/should support an abstract base but I don't think that's worth the investment right now.

tkaemming · 2023-12-18T21:42:15Z

posthog/temporal/batch_exports/squash_person_overrides.py

+DEFAULT_POSTGRES_PERSON_OVERRIDES_MANAGER = "mappings"
+assert DEFAULT_POSTGRES_PERSON_OVERRIDES_MANAGER in POSTGRES_PERSON_OVERRIDES_MANAGERS


Also clunky, but similar to above I don't think it's worth wrapping this in an abstraction.

tkaemming · 2023-12-18T21:45:17Z

posthog/temporal/batch_exports/squash_person_overrides.py

@@ -184,6 +421,7 @@ class QueryInputs:
    dictionary_name: str = "person_overrides_join_dict"
    dry_run: bool = True
    _latest_created_at: str | datetime | None = None
+    postgres_person_overrides_manager: str = DEFAULT_POSTGRES_PERSON_OVERRIDES_MANAGER


I don't really like tacking this on here as it's only used in one activity within the workflow.

You could have a class holding all the workflow activities and also the manager. This class and the manager would be initialized once when starting the temporal worker. Here's a sample of the pattern: https://github.com/temporalio/samples-python/blob/main/hello/hello_activity_method.py

Not saying it would be better, just wanted to give a suggestion given your comment 👍

tkaemming · 2023-12-18T21:49:52Z

posthog/temporal/tests/test_squash_person_overrides_workflow.py

@@ -928,7 +926,7 @@ def team_id(query_inputs, organization_uuid, pg_connection):
                """,
                {"organization_uuid": organization_uuid},
            )
-            team_id = cursor.fetchone()
+            [team_id] = cursor.fetchone()


Was honestly kind of surprised this was working at all — this is passed back into psycopg2.execute as parameters later on in situations where I'd assume it'd expect a scalar, but I guess this works?

tkaemming · 2023-12-18T21:52:14Z

posthog/temporal/tests/test_squash_person_overrides_workflow.py

+    # XXX: Several activity-based tests use this person overrides fixture and
+    # should vary their behavior to ensure that they work with both the old
+    # (mappings) and new (flat) approaches, but not all tests that use
+    # `query_inputs` need to be vary on the overrides manager type as many of
+    # them don't use Postgres overrides at all. To ensure that whenever Postgres
+    # overrides *are* used, we need to update the fixture here. This indirection
+    # isn't good, but this code should be short-lived, right? (... right???)
+    query_inputs.postgres_person_overrides_manager = request.param


There are definitely better approaches here, but I couldn't think of any that wouldn't require a bunch of refactoring. Second opinions welcomed!

tkaemming · 2023-12-18T21:54:51Z

posthog/temporal/batch_exports/squash_person_overrides.py

+        overrides_manager = inputs.get_postgres_person_overrides_manager(connection)
+        for person_override_to_delete in inputs.iter_person_overides_to_delete():
+            activity.logger.debug("%s", person_override_to_delete)
+            overrides_manager.delete(person_override_to_delete, inputs.dry_run)


@bretthoerner brought up a good question here about whether we should bother with deleting this data at all: #19220 (comment)

I'm not sure I have a strong opinion one way or another.

In the case of data corruption, our best source of truth for the event to person relationship would be by looking up the current person associated with an event's distinct_id¹, so I'm not sure we gain a lot by keeping these records around. (I figure that if we break persons somehow, we likely break them by accidentally writing bad data to this table.)

If we expect to need to rebuild the ClickHouse overrides table for some reason though (schema change, mostly, or the unlikely scenario that the Kafka engine goes haywire), I could see the benefit of keeping this around, but I'm not sure how likely that would be.

Footnotes

h/t @tiina303 for pointing this out to me a few weeks ago! ↩

bretthoerner · 2023-12-19T18:33:30Z

posthog/temporal/batch_exports/squash_person_overrides.py

+                    %(team_id)s,
+                    %(old_person_id)s,
+                    %(override_person_id)s,
+                    NOW(),


Because this is only for tests, right? Otherwise we'd use something from one of the users?

bretthoerner · 2023-12-19T18:34:08Z

posthog/temporal/batch_exports/squash_person_overrides.py

+                    %(team_id)s,
+                    %(old_person_id)s,
+                    %(override_person_id)s,
+                    NOW(),


Same question here.

Yep, same here.

bretthoerner · 2023-12-19T18:34:57Z

posthog/temporal/tests/test_squash_person_overrides_workflow.py

@@ -928,7 +926,7 @@ def team_id(query_inputs, organization_uuid, pg_connection):
                """,
                {"organization_uuid": organization_uuid},
            )
-            team_id = cursor.fetchone()
+            [team_id] = cursor.fetchone()


bretthoerner · 2023-12-19T18:36:01Z

posthog/temporal/batch_exports/squash_person_overrides.py

+    override_person_id: UUID
+
+
+class PostgresPersonOverridesManager:


No strong opinion here. PostgresPersonOverrideStrategy is good too. 🤷‍♂️

posthog-bot · 2023-12-27T07:30:57Z

This PR hasn't seen activity in a week! Should it be merged, closed, or further worked on? If you want to keep it open, post a comment or remove the stale label – otherwise this will be closed in another week.

posthog/temporal/batch_exports/squash_person_overrides.py

…sh workflow (#19347)

* master: (94 commits) feat: Add cta on billing for >20k annual spend customers (#19508) refactor(temporal/squash): Support flat person override table in squash workflow (#19347) fix(surveys): remove link in user interview template (#19584) feat: populate plugin capabilities on install and edit (#19188) chore: Make plugin-server ignore deleted plugin configs (#18444) chore: Add Flutter feature flags snippets (#19563) fix(bi): fixed some of the query duplications (#19573) fix: assume typeless series nodes are of type events node (#19550) chore(deps): bump chromaui/action from 1 to 10 (#19560) fix(trends): fix breakdowns persons label (#19534) fix(trends): fix breakdown legend (#19533) feat: incremental updates for mobile transformer (#19567) chore(deps): bump peter-evans/find-comment from 1 to 2 (#19559) chore(data-warehouse): cleanup unused celery code and extend time (#19568) Revert "feat(data-warehouse): hubspot integration" (#19569) feat(data-warehouse): hubspot integration (#19529) feat: Feature gate session replay controls using available_product_features (#19401) fix: Padding on bullet lists (#19565) chore: Post 3000 LemonButtton cleanup (#19540) chore(deps): bump aws-actions/amazon-ecr-login from 1 to 2 (#19558) ...

tkaemming mentioned this pull request Dec 15, 2023

refactor(plugin-server): Add flat person override table and writer #19220

Merged

Base automatically changed from poe-deferred-override-worker-flat to master December 18, 2023 20:11

tkaemming added 15 commits December 18, 2023 12:21

Start to split out Postgres override handling into its own abstraction.

5a61365

Start to modify test fixtures.

77e3473

Unpack team ID fixture row into scalar value.

5273918

Generalize tests where applicable to handle both modes.

3d30396

Inline queries in applicable functions

842d53a

Add FlatPostgresPersonOverridesManager

8af8603

A working yet very questionable integration into the workflow and tests

78f8426

Clean up some names and organization.

0d5634a

Comment gross test behavior

130a578

Tag the person override fixtures with a type

88074cc

Tidy up tests

ddc4e37

Consolidate override data structures

b78e686

Remove some unused query input fixtures

36a30ac

Clean up more names

75ad916

Fix typing in tests

7d04c33

tkaemming force-pushed the poe-deferred-override-worker-flat-squash branch from f24a57d to 7d04c33 Compare December 18, 2023 20:21

tkaemming commented Dec 18, 2023

View reviewed changes

tkaemming requested a review from a team December 18, 2023 22:11

tkaemming marked this pull request as ready for review December 18, 2023 22:13

bretthoerner approved these changes Dec 19, 2023

View reviewed changes

posthog-bot added the stale label Dec 27, 2023

tkaemming removed the stale label Dec 28, 2023

Merge branch 'master' into poe-deferred-override-worker-flat-squash

cb48ede

tkaemming mentioned this pull request Jan 1, 2024

test(async-migrations): Change timestamp upper bound to account for it being 2024 #19544

Merged

tomasfarias reviewed Jan 2, 2024

View reviewed changes

posthog/temporal/batch_exports/squash_person_overrides.py Show resolved Hide resolved

tkaemming added 2 commits January 2, 2024 16:02

Merge branch 'master' into poe-deferred-override-worker-flat-squash

7bf8fde

Use flat override path by default, see PostHog/charts#623

5e177b9

tkaemming merged commit 07785b2 into master Jan 3, 2024
72 checks passed

tkaemming deleted the poe-deferred-override-worker-flat-squash branch January 3, 2024 23:14

fuziontech pushed a commit that referenced this pull request Jan 4, 2024

refactor(temporal/squash): Support flat person override table in squa…

ab4d0eb

…sh workflow (#19347)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(temporal/squash): Support flat person override table in squash workflow #19347

refactor(temporal/squash): Support flat person override table in squash workflow #19347

tkaemming commented Dec 15, 2023 •

edited

Loading

tkaemming Dec 18, 2023

tkaemming Dec 18, 2023

tkaemming Dec 18, 2023

bretthoerner Dec 19, 2023

tkaemming Dec 18, 2023

tkaemming Dec 18, 2023

tkaemming Dec 18, 2023

tomasfarias Jan 2, 2024

tomasfarias Jan 2, 2024

tkaemming Dec 18, 2023 •

edited

Loading

bretthoerner Dec 19, 2023

tkaemming Dec 18, 2023

tkaemming Dec 18, 2023

bretthoerner Dec 19, 2023

tkaemming Jan 1, 2024

bretthoerner Dec 19, 2023

tkaemming Jan 1, 2024

bretthoerner Dec 19, 2023

bretthoerner Dec 19, 2023

posthog-bot commented Dec 27, 2023

		override_person_id: UUID


		class PostgresPersonOverridesManager:

		DEFAULT_POSTGRES_PERSON_OVERRIDES_MANAGER = "mappings"
		assert DEFAULT_POSTGRES_PERSON_OVERRIDES_MANAGER in POSTGRES_PERSON_OVERRIDES_MANAGERS

refactor(temporal/squash): Support flat person override table in squash workflow #19347

refactor(temporal/squash): Support flat person override table in squash workflow #19347

Conversation

tkaemming commented Dec 15, 2023 • edited Loading

Problem

How did you test this code?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tkaemming Dec 18, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Footnotes

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

posthog-bot commented Dec 27, 2023

tkaemming commented Dec 15, 2023 •

edited

Loading

tkaemming Dec 18, 2023 •

edited

Loading