Skip to content

chore(warehouse-sources): extract DeltaWriter from DeltaTableHelper #76084

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
estefaniarabadan wants to merge 1 commit into
base: master
Choose a base branch
from
Open

chore(warehouse-sources): extract DeltaWriter from DeltaTableHelper #76084

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion posthog/temporal/data_modeling/activities/materialize_view.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -505,7 +505,7 @@ async def materialize_view_activity(inputs: MaterializeViewInputs) -> Materializ
pa_schema: pa.Schema | None = None

# write each batch as its own delta commit, imitating the data_imports pipeline
# (DeltaTableHelper.write_to_deltalake): the first batch overwrites — creating the
# (DeltaWriter.write): the first batch overwrites — creating the
# table from the exact arrow schema, which pins column case like `personId` — and
# later batches append with schema_mode="merge". this keeps peak memory at ~one
# batch (hogql_table yields ~100MB combined batches) and, because each write is a
Expand Down
3 changes: 2 additions & 1 deletion products/warehouse_sources/backend/temporal/data_imports/pipelines/common/load.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,7 @@ async def _seed_cdc_companion_from_snapshot(
SCD2_VALID_FROM_COLUMN,
SCD2_VALID_TO_COLUMN,
)
from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta.writer import DeltaWriter
from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import (
DeltaTableHelper,
)
Expand Down Expand Up @@ -242,7 +243,7 @@ def _read_next_batch(r: pa.RecordBatchReader) -> pa.RecordBatch | None:

# Plain append — the companion table is freshly reset so there are no existing
# rows to close, making SCD2 merge unnecessary.
await companion_helper.write_to_deltalake(
await DeltaWriter(companion_helper).write(
data=batch_table,
write_type="append",
should_overwrite_table=False,
Expand Down
6 changes: 3 additions & 3 deletions ...ts/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta/test/test_scd2.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
import pyarrow.compute as pc

from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta.scd2 import Scd2DeltaWriter
from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta_table_helper import DeltaTableHelper
from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.delta.writer import commit_matches
from products.warehouse_sources.backend.temporal.data_imports.pipelines.core.test.test_delta_table_helper import (
_decimal_array,
_make_local_helper,
Expand Down Expand Up @@ -95,8 +95,8 @@ async def test_only_terminal_append_commit_carries_metadata(self, tmp_path: Path
await _make_writer(delta_path).write(data=batch, primary_keys=["id"], commit_metadata=metadata)

history = deltalake.DeltaTable(delta_path).history()
# Newest first: [append (WRITE), close (MERGE), seed (WRITE)]. _commit_matches is the
# Newest first: [append (WRITE), close (MERGE), seed (WRITE)]. commit_matches is the
# same layout-agnostic check has_batch_been_committed uses for redelivery dedup.
tagged = [c["operation"] for c in history if DeltaTableHelper._commit_matches(c, metadata)]
tagged = [c["operation"] for c in history if commit_matches(c, metadata)]
assert tagged == ["WRITE"]
assert any(c["operation"] == "MERGE" for c in history)
Loading
Loading