feat(warehouse): structured deltalite write stats + duration

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}")

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,16 +1423,26 @@ 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):
             assert await self._call(self._helper()) is False
 
     @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):

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)",
+)