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..d78f234a 100644 --- a/tests/test_spill.py +++ b/tests/test_spill.py @@ -697,11 +697,15 @@ 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 + # 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: @@ -783,6 +787,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}