source-posthog: turn persons into an incremental stream

[nicolaslazo]

Description:

PostHog's persons stream is a snapshot type, which is currently filling up storage space in some reactors as it tries to accumulate thousands of documents in memory without checkpointing its progress.

This PR implements secondary cursor support for HogQL queries. This allows us to rely on last_seen_at for update timestamps, falling back to created_at when not available.

Workflow steps:

(How does one use this feature, and how has it changed)

Documentation links affected:

(list any documentation links that you created, or existing ones that you've identified as needing updates, along with a brief description)

Notes for reviewers:

Tested locally through flowctl preview.

[Alex-Bair]

Had a few items to address before approving. This comment about adding new projects to the Persons resource state looks like it also applies to the Events stream. I missed that when I reviewed the connector initially, but we'll want to improve that handling soon, either in this PR or in a follow up one.

[Alex-Bair]

_meta/project_id is part of the composite key, and we shouldn't allow it to be nullable in order to make it easier to materialize Persons to an SQL destination. If we did have nullable keys, SQL materializations wouldn't be able to materialize Persons out of the box since SQL destinations reject nullable primary keys.

[Alex-Bair]

Thinking through an edge case, what happens if there is a long run of records all with the same cursor value? Is it possible that we the following could happen?

• We fetch the first handful of records at cursor N.
• We stop fetching the remaining records at cursor N because the BACKFILL_TIMEOUT_PERIOD has elapsed.
• We yield N as the next cursor value.
• We reinvoke backfill_persons and have it start fetching records with cursor values greater than N. Meaning, we miss the remaining records at cursor N that would have taken more than BACKFILL_TIMEOUT_PERIOD to fetch.

That looks possible to me, but let me know if you disagree. I've handled the above situation in other connectors by only breaking out of the pagination loop early when both BACKFILL_TIMEOUT_PERIOD has elapsed and the next document's cursor value is greater than the previous document's cursor value. Here's an example from source-klaviyo-native where I implemented that logic:

connectors/source-klaviyo-native/source_klaviyo_native/api/common.py

Lines 299 to 312 in 5bf47b1

	# If we see a document updated more recently than the previous
	# document we emitted, checkpoint the previous documents.
	if (
	count >= CHECKPOINT_INTERVAL and
	doc_cursor > last_seen_dt
	):
	yield dt_to_str(last_seen_dt)
	count = 0
	# If backfill_incremental_resources has been running for more than TARGET_FETCH_PAGE_INVOCATION_RUN_TIME
	# minutes, then yield control back to the CDK after a checkpoint. This forces backfills to check the
	# stopping event every so often & gracefully exit if it's set.
	if time.time() - start_time >= TARGET_FETCH_PAGE_INVOCATION_RUN_TIME:
	return

[nicolaslazo]

I've already had to tackle a similar issue in source-quickbooks. Like mentioned on our 1:1, I'll try to extract the checkpointing logic to a decorator abstract enough it could become a CDK feature 👍

[Alex-Bair]

That backfill_timeout decorator is interesting! It looks like it may be general enough to use elsewhere as long as the item's cursor value is a datetime that's exposed via the get_cursor method. It may also be worth having a comment or docstring stating that the decorator should only be used when the backfill_fn yields documents in ascending order, just so it's easier for us to remember the specific situations this decorator should be used in. Let's monitor how it does, and if we like it, we can figure out how to abstract it into the CDK somehow.

[Alex-Bair]

LGTM % small nit of regenerating the schema.yaml files with flowctl raw discover to reflect that project_id is now non-nullable.

Note: We'll need to perform dataflow resets for all production captures' Persons binding due to the change in the key field.