Spill reduce/read profiling split — #280 step 0 (measurement enablement)

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.

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
 

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}