diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py index ea8bd5a14371..8139924b3a2e 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/delta_table_helper.py @@ -1,4 +1,5 @@ import json +import time import asyncio import contextlib from collections.abc import Callable, Sequence @@ -303,6 +304,27 @@ def _merge_predicate_ops(normalized_primary_keys: list[str]) -> list[str]: return [f"(source.{c} IS NOT DISTINCT FROM target.{c})" for c in normalized_primary_keys] +def _deltalite_write_stats(stats: Any) -> dict[str, int | float | str | bool]: + """Flatten a deltalite ``UpsertStats`` into scalar log fields for structured, parseable output. + + Enumerates the object's public scalar attributes (the pyo3 ``#[pyo3(get)]`` getters — version, + partitions_touched, files_added/removed/carried_over/probed, rows_updated/inserted/copied, + source_rows, null_pk_rows, …) rather than a fixed list, so fields added crate-side later (e.g. + per-phase timings) surface automatically. Best-effort — a stats change must never break the write. + """ + fields: dict[str, int | float | str | bool] = {} + for name in dir(stats): + if name.startswith("_"): + continue + try: + value = getattr(stats, name) + except Exception: # noqa: BLE001 - a flaky getter must not break logging a committed write + continue + if isinstance(value, bool | int | float | str): + fields[name] = value + return fields + + def delta_storage_options() -> dict[str, str]: """delta-rs storage options for the data-warehouse bucket, independent of any import job — so a read path (e.g. the person-property backfill) can open a Delta table without constructing a full @@ -587,6 +609,7 @@ async def _write_via_deltalite( import deltalite from products.warehouse_sources.backend.temporal.data_imports.pipelines.pipeline_v3.load.metrics import ( + DELTALITE_WRITE_DURATION_SECONDS, DELTALITE_WRITE_TOTAL, ) @@ -603,7 +626,9 @@ def _upsert() -> Any: commit_metadata=commit_metadata, ) + started = time.perf_counter() stats = await asyncio.to_thread(_upsert) + duration_s = time.perf_counter() - started except Exception as e: # noqa: BLE001 - pre-commit failure: nothing committed, fall back to MERGE await self._logger.awarning( f"deltalite write failed; falling back to delta-rs MERGE (sync unaffected): {e}" @@ -622,11 +647,17 @@ def _upsert() -> Any: # Refresh the in-memory delta-rs handle to deltalite's new version so the table returned by # write_to_deltalake (and any subsequent reads) reflects the real state. await asyncio.to_thread(existing_delta_table.update_incremental) + # Structured, parseable stats (parity with the old `Delta Merge Stats: {json}` line): every + # UpsertStats field becomes its own log key, plus the wall-clock duration. `_deltalite_write_stats` + # enumerates the pyo3 getters, so fields added crate-side later (e.g. per-phase timings) flow + # through here without a code change. await self._logger.ainfo( - f"deltalite write: committed v{stats.version} " - f"(+{stats.rows_inserted} inserted / ~{stats.rows_updated} updated / {stats.rows_copied} copied)" + "deltalite write: committed", + duration_ms=round(duration_s * 1000), + **_deltalite_write_stats(stats), ) DELTALITE_WRITE_TOTAL.labels(outcome="written").inc() + DELTALITE_WRITE_DURATION_SECONDS.observe(duration_s) except Exception as e: # noqa: BLE001 - the write is committed; bookkeeping must never raise with contextlib.suppress(Exception): await self._logger.awarning(f"deltalite write committed but post-commit bookkeeping failed: {e}") diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py index c456a396dc7b..cb535a734b1d 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/core/test/test_delta_table_helper.py @@ -2,6 +2,7 @@ from datetime import UTC, datetime from decimal import Decimal from pathlib import Path +from types import SimpleNamespace from typing import Any, cast import pytest @@ -24,6 +25,7 @@ DELTA_MERGE_CONFLICT_RETRIES, DeltaTableHelper, _delta_merge_spill_kwargs, + _deltalite_write_stats, _first_per_pk_table, _merge_predicate_ops, _realign_decimal_buffers, @@ -1421,6 +1423,12 @@ async def test_skips_without_primary_keys(self): assert wrote is False flag.assert_not_called() + def test_write_stats_flattens_scalar_getters_only(self): + # Enumerates scalar attributes (so future crate fields flow through) and drops methods/non-scalars. + stats = SimpleNamespace(version=7, rows_inserted=2, files_added=1, _private=9) + stats.helper = lambda: None # callable attribute must be ignored + assert _deltalite_write_stats(stats) == {"version": 7, "rows_inserted": 2, "files_added": 1} + @pytest.mark.asyncio async def test_falls_back_when_flag_disabled(self): with patch(self._FLAG, return_value=False): @@ -1428,9 +1436,13 @@ async def test_falls_back_when_flag_disabled(self): @pytest.mark.asyncio async def test_writes_via_deltalite_when_enabled(self): - helper = self._helper() + logger = _make_logger() # captured so we can inspect the structured log without hitting the typed attr + helper = DeltaTableHelper(resource_name="t", job=MagicMock(team_id=2, schema_id="sch-1"), logger=logger) existing = MagicMock() - fake_stats = MagicMock(version=5, rows_inserted=3, rows_updated=2, rows_copied=10) + # SimpleNamespace stands in for the pyo3 UpsertStats: predictable scalar getters for the structured log. + fake_stats = SimpleNamespace( + version=5, partitions_touched=1, rows_inserted=3, rows_updated=2, rows_copied=10, null_pk_rows=0 + ) fake_table = MagicMock() fake_table.upsert.return_value = fake_stats fake_deltalite = MagicMock() @@ -1454,6 +1466,13 @@ async def test_writes_via_deltalite_when_enabled(self): # PARTITION_KEY is passed as the partition arg when the table is partitioned. assert fake_table.upsert.call_args.args[2] == PARTITION_KEY existing.update_incremental.assert_called_once() + # The commit is logged with the UpsertStats fields as structured keys + a duration, so it's parseable. + logger.ainfo.assert_called_once() + log_kwargs = logger.ainfo.call_args.kwargs + assert log_kwargs["version"] == 5 + assert log_kwargs["rows_inserted"] == 3 + assert log_kwargs["partitions_touched"] == 1 + assert "duration_ms" in log_kwargs @pytest.mark.asyncio async def test_falls_back_when_deltalite_raises(self): diff --git a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/load/metrics.py b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/load/metrics.py index af0701cc5b29..823f82034474 100644 --- a/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/load/metrics.py +++ b/products/warehouse_sources/backend/temporal/data_imports/pipelines/pipeline_v3/load/metrics.py @@ -55,3 +55,8 @@ "Incremental merges routed to deltalite's real write path, by outcome", labelnames=["outcome"], ) + +DELTALITE_WRITE_DURATION_SECONDS = Histogram( + "warehouse_load_deltalite_write_duration_seconds", + "Wall-clock time of a deltalite real write (DeltaLiteTable.upsert)", +)