From d5b24bd65ae6f5981c3d82d2142a2ca2b0e7fbe4 Mon Sep 17 00:00:00 2001 From: espg Date: Sat, 18 Jul 2026 15:30:36 -0700 Subject: [PATCH 1/2] instrument spill reduce-CPU vs read-I/O split (issue #280) --- src/zagg/processing/spill.py | 7 +++++++ src/zagg/processing/worker.py | 3 +++ tests/test_spill.py | 14 ++++++++++++++ 3 files changed, 24 insertions(+) diff --git a/src/zagg/processing/spill.py b/src/zagg/processing/spill.py index 2d79e7d0..b1ac05dc 100644 --- a/src/zagg/processing/spill.py +++ b/src/zagg/processing/spill.py @@ -507,6 +507,11 @@ def __init__( self.spill_bytes = 0 self.spill_write_s = 0.0 self.spill_read_s = 0.0 + # Wall spent in the block fold's *compute* (group + build + merge), + # separate from spill_read_s (the read-back I/O), so a profile run can + # split reduce-CPU vs read-I/O — the measurement that decides whether the + # #280 reducer parallelization should target compute or I/O. + self.spill_reduce_s = 0.0 self._buffer: list = [] self._buffered_granules = 0 @@ -637,6 +642,7 @@ def _fold_block(self, block: SpillBlock) -> None: t0 = time.perf_counter() cells, cols = block.read_partition(key, close=True) self.spill_read_s += time.perf_counter() - t0 + t1 = time.perf_counter() col_arrays, cell_to_slice = _group_columns(cols, cells) del cells, cols for cell, (start, end) in cell_to_slice.items(): @@ -650,6 +656,7 @@ def _fold_block(self, block: SpillBlock) -> None: ) else: self._digest_parts[name].setdefault(cell, []).append(fresh) + self.spill_reduce_s += time.perf_counter() - t1 def _finalize_kway(self) -> None: """Collapse each k-way field's per-block parts into one digest per cell. diff --git a/src/zagg/processing/worker.py b/src/zagg/processing/worker.py index f8c5ae55..dda1a8be 100644 --- a/src/zagg/processing/worker.py +++ b/src/zagg/processing/worker.py @@ -714,6 +714,9 @@ def _iter_granule_reads(): # closes mid-read) or the aggregate phase (single-block reduce). phase_timings["spill_write_s"] = buffered.spill_write_s phase_timings["spill_read_s"] = buffered.spill_read_s + # Block-fold compute (group+build+merge), split from read-back I/O so + # a profile run shows reduce-CPU vs read-I/O — the #280 measurement. + phase_timings["spill_reduce_s"] = buffered.spill_reduce_s phase_timings["spill_bytes"] = buffered.spill_bytes metadata["phase_timings"] = phase_timings diff --git a/tests/test_spill.py b/tests/test_spill.py index 1398b646..e9dacd45 100644 --- a/tests/test_spill.py +++ b/tests/test_spill.py @@ -697,11 +697,13 @@ def test_profile_carries_spill_instrumentation(self, monkeypatch): "aggregate", "spill_write_s", "spill_read_s", + "spill_reduce_s", "spill_bytes", } assert timings["spill_bytes"] > 0 assert timings["spill_write_s"] >= 0 assert timings["spill_read_s"] >= 0 + assert timings["spill_reduce_s"] >= 0 class TestSpillWorkerMultiBlock: @@ -783,6 +785,18 @@ def test_standard_kway_multi_block_tracks_pooled_quantiles(self, monkeypatch): for q in (0.1, 0.5, 0.9): assert abs(quantile_from_tdigest(ds, q) - quantile_from_tdigest(dp, q)) < 1.0 + def test_multi_block_reduce_time_is_captured(self, monkeypatch): + # The block fold runs only in the multi-block regime, so spill_reduce_s + # (the #280 reduce-CPU-vs-read-I/O split) is populated here, not on the + # single-block exact path. + self._force_tiny_blocks(monkeypatch) + key = _shard_key() + cfg = _config(streaming={"buffer_granules": 2, "mode": "spill"}) + grid = _grid(cfg) + dfs = _granule_dfs(grid, key, _CELL_LISTS, obs_per_cell=200, seed=3) + _, _, meta = _run(monkeypatch, cfg, grid, key, dfs, profile=True) + assert meta["phase_timings"]["spill_reduce_s"] > 0 + def test_multi_block_actually_engaged_and_counts_exact(self, monkeypatch): self._force_tiny_blocks(monkeypatch) closes = {"n": 0} From aba9fa79f131462331ced0f9186d0f7a21c6a604 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 10:16:35 +0000 Subject: [PATCH 2/2] fold review: pin single-block spill_reduce_s == 0.0 (issue #280) --- tests/test_spill.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_spill.py b/tests/test_spill.py index e9dacd45..d78f234a 100644 --- a/tests/test_spill.py +++ b/tests/test_spill.py @@ -703,7 +703,9 @@ def test_profile_carries_spill_instrumentation(self, monkeypatch): assert timings["spill_bytes"] > 0 assert timings["spill_write_s"] >= 0 assert timings["spill_read_s"] >= 0 - assert timings["spill_reduce_s"] >= 0 + # Single-block regime never enters _fold_block, so the reduce-CPU split + # is deterministically zero here (the multi-block case asserts > 0). + assert timings["spill_reduce_s"] == 0.0 class TestSpillWorkerMultiBlock: