refactor: Snowflake batch export

[tomasfarias]

Problem

Snowflake is the first destination we supported for PostHog batch exports, as such the code has been missing some best practices and features we introduced in destinations implemented later, but never ported back to Snowflake.

Changes

• Wrap the Snowflake API to make it async.
  • Which allows us to heartbeat...
• Implement Heartbeat on progress to allow resuming from partial exports.
  • And now that we heartbeat we can...
• Implement resume based on last heartbeat progress values.

This heartbeat API is going to be rolled out to all destinations, which is why I've dropped it in a utilities module. Long term, I would like to shuffle modules around and end up with a file structure like:

. batch_exports
├── __main__.py (start_worker() with all workflows and activities in this package)
├── activities (common activities shared by all workflows)
│   ├── create_batch_export.py
│   └── ...
├── workflows
│   ├── base.py (PostHogWorkflow base class)
│   ├── bigquery_batch_export_workflow.py
│   ├── snowflake_batch_export_workflow.py
│   └── ...
├── models (or maybe just a models.py)
│   ├── batch_export_model.py
│   ├── backfill_batch_export_model.py
│   └── ...
├── utilities (if any of these gets too big, we can move them to their own separate module)
│   ├── heartbeat.py
│   ├── temporary_file.py
│   ├── clickhouse.py
│   ├── service.py
│   └── ...
├── temporal
│   ├── client.py
│   ├── codec.py
│   ├── worker.py
│   └── ...
└── tests
    ├── workflows
    │   ├── test_bigquery_batch_export_workflow.py
    │   └── ...
    ├── temporal
    │   ├── test_codec.py
    │   └── ...
    └── utilities
        ├── test_clickhouse.py
        ├── test_heartbeat.py
        └── ...

The ideas behind a refactoring like this are:

1. Separate batch exports from the rest of the PostHog monolith.
   a. If we eventually want batch exports to be separate from PostHog, for starters, it has to be in it in its own directory.
   b. This also makes it easy to eventually refactor the models to avoid any dependencies to PostHog.
2. Temporal is just the tool we use for the job, batch exports is the actual application. So, batch exports should be the root dir.
   a. Currently, things being separated between batch_exports/ and temporal/ makes no sense.
   b. But it also doesn't make sense to fully integrate into PostHog if long term we want batch exports to be a separate package
   c. The only exception is the batch_exports/http.py module which will remain part of PostHog.

This PR also has some other non-heartbeat related changes:

• Use the BatchExportTemporaryFile class to manage file writing.
• Add optional unit tests like those implemented for BigQuery/Redshift.

👉 Stay up-to-date with PostHog coding conventions for a smoother review.

How did you test this code?

New unit tests ran against real Snowflake:

$  SNOWFLAKE_USERNAME="username" SNOWFLAKE_ACCOUNT="account" SNOWFLAKE_PASSWORD="password" SNOWFLAKE_WAREHOUSE="warehouse" DEBUG=1 pytest posthog/temporal/tests/batch_exports/test_snowflake_batch_export_workflow.py::test_snowflake_export_workflow -vv
===================================================== test session starts =====================================================
platform linux -- Python 3.10.10, pytest-7.4.0, pluggy-0.13.1 -- src/github.com/PostHog/posthog/.direnv/python-3.10.10/bin/python
cachedir: .pytest_cache
django: settings: posthog.settings (from ini)
rootdir: src/github.com/PostHog/posthog
configfile: pytest.ini
plugins: asyncio-0.21.1, icdiff-0.6, flaky-3.7.0, env-0.8.2, Faker-17.5.0, syrupy-1.7.4, mock-3.11.1, split-0.8.1, django-4.5.2, cov-4.1.0
asyncio: mode=strict
collected 4 items                                                                                                             

posthog/temporal/tests/batch_exports/test_snowflake_batch_export_workflow.py::test_snowflake_export_workflow[None-hour] PASSED [ 25%]
posthog/temporal/tests/batch_exports/test_snowflake_batch_export_workflow.py::test_snowflake_export_workflow[None-day] PASSED [ 50%]
posthog/temporal/tests/batch_exports/test_snowflake_batch_export_workflow.py::test_snowflake_export_workflow[exclude_events1-hour] PASSED [ 75%]
posthog/temporal/tests/batch_exports/test_snowflake_batch_export_workflow.py::test_snowflake_export_workflow[exclude_events1-day] PASSED [100%]

--------------------------------------------------- snapshot report summary ---------------------------------------------------

===================================================== 4 passed in 35.77s ======================================================

[tomasfarias]

We should hold off on merging this one until #18467 is merged due to potential conflicts.

[bretthoerner]

There's a conflict from my merge, but LGTM.

I like the plans to extract batch exports to their own directory/app/service/image, no real comments there at this point.

[bretthoerner]

Should this include the exception itself?

[tomasfarias]

Ah, yeah, this log should be promoted to ERROR level and use logger.exception.

[bretthoerner]

Will this exception end up in some existing log at least?

[tomasfarias]

It will bubble up and the activity will fail, so we will see it in the temporal error logs.

[tomasfarias]

The TODO was just thinking if we should do anything more, maybe log ourselves to add some context. I'll do that at least once we merge the structlog PR and I can rebase this one.

[bretthoerner]

Totally aside, it's wild to me that their execute fn does or doesn't work only with certain statements. Weird!

[tomasfarias]

There is some (very little) explanation of why here: snowflakedb/snowflake-connector-python#1227 (comment).

[bretthoerner]

Is the unique filename thing true for the lifetime of the table? Do we need to worry at all about a temporary file name (which is I think what we're using here) being reused?

[tomasfarias]

If COPY is successful after we are done uploading everything, the files will be purged, so their names become available again.

That being said, the purge can fail (even without COPY failing), or we could fail somewhere before COPY. In that case, we are hoping that Python will not generate the same name for our temporary file again. First, Python generates an infinite sequence of random names (with this: https://github.com/python/cpython/blob/3.10/Lib/tempfile.py#L132), and every time it needs a name for a temp file, it calls next to get the next randomly generated name (here: https://github.com/python/cpython/blob/3.10/Lib/tempfile.py#L252), and uses that as the file name (here: https://github.com/python/cpython/blob/3.10/Lib/tempfile.py#L556).

So, worst possible scenario, our protection from collisions boils down to how likely or under what circumstances will the Python RNG choose the same sequence of 8 characters. We are not manually seeding it for any reason, so I think we should be safe and that if it ever happens it will be a good anecdote.

EDIT: I ran through the name generation process in this comment as I was keeping notes as I looked up how it worked just now. Not my intention to imply I know everything or sound pedantic!

[bretthoerner]

Thanks for all the detail!

[github-actions]

Size Change: -2.64 kB (0%)

Total Size: 2.01 MB

Filename	Size	Change
frontend/dist/toolbar.js	2.01 MB	-2.64 kB (0%)

_{compressed-size-action}