perf: close the build-time gap, and measure the mask the ladder never measured

bench/README.md

@@ -67,9 +67,14 @@ verdict off the SQL"), not to cover the language:
 | case | shape | why |
 |---|---|---|
 | `dispatch` | pointwise bounds + one `sum` per row | raw throughput, and the case a dense eager broadcast is best at — so our worst ratio |
+| `sparse` | the same math behind a 2-D `where` | the mask itself: row absence against NaN-padding. `dispatch`'s `where: p_max > 0` removes *nothing* — the generated p_max is always positive — so without this case the ladder only ever measured a dense coord product |
 | `transport` | three `group_sum` joins per row | the mapping-table path, where the eager lane must materialise a bus x generator product |
 
-Both scale on snapshots; `bench/cases.py` holds the ladders. Data is generated
+`sparse` also earns its keep on our side of the fence: its mask reads the
+leading dim, which puts the label frame on the general sorted path rather than
+the arithmetic one, so the ladder covers both.
+
+All three scale on snapshots; `bench/cases.py` holds the ladders. Data is generated
 deterministically (a blake2b digest of the shape seeds the RNG — `hash()` is
 salted per process and would give the two arms different numbers), cached under
 `bench/.cache/`, and feasible by construction: every bus can serve its own load

bench/cases.py

@@ -138,6 +138,62 @@ def _dispatch_eager(paths: dict[str, str]) -> tuple[dict[str, Any], dict[str, An
     return data, coords
 
 
+# --------------------------------------------------------------------------
+# sparse — the same dispatch math with a mask that actually removes rows
+
+#: Fraction of the (snapshot, generator) product `available` keeps. Low enough
+#: that row absence and NaN-padding are doing visibly different work; high
+#: enough that every snapshot can still meet its load.
+SPARSE_DENSITY = 0.2
+
+
+def _sparse_data(shape: Shape, dest: Path) -> dict[str, str]:
+    rng = _seed(shape)
+    n_snap, n_gen = shape.sizes['snapshot'], shape.sizes['generator']
+    gens = [f'g{i:05d}' for i in range(n_gen)]
+
+    p_max = rng.uniform(50.0, 150.0, n_gen)
+    cost = rng.uniform(10.0, 100.0, n_gen)
+    available = (rng.random((n_snap, n_gen)) < SPARSE_DENSITY).astype(float)
+    # every snapshot needs at least one generator, or the model is infeasible
+    # and the parity gate cannot compare two objectives
+    available[np.arange(n_snap), rng.integers(0, n_gen, n_snap)] = 1.0
+
+    # load against the capacity actually reachable in the tightest snapshot,
+    # so feasibility does not depend on the draw
+    reachable = (available * p_max).sum(axis=1)
+    load = reachable.min() * 0.5 * (0.8 + 0.4 * rng.random(n_snap))
+
+    return _dump(
+        {
+            'available': pd.DataFrame(
+                {
+                    'snapshot': np.repeat(np.arange(n_snap), n_gen),
+                    'generator': np.tile(gens, n_snap),
+                    'value': available.reshape(-1),
+                }
+            ),
+            'p_max': pd.DataFrame({'generator': gens, 'value': p_max}),
+            'cost': pd.DataFrame({'generator': gens, 'value': cost}),
+            'load': pd.DataFrame({'snapshot': np.arange(n_snap), 'value': load}),
+        },
+        dest,
+    )
+
+
+def _sparse_eager(paths: dict[str, str]) -> tuple[dict[str, Any], dict[str, Any]]:
+    p_max = pd.read_parquet(paths['p_max']).set_index('generator')['value']
+    cost = pd.read_parquet(paths['cost']).set_index('generator')['value']
+    load = pd.read_parquet(paths['load']).set_index('snapshot')['value']
+    available = pd.read_parquet(paths['available']).set_index(['snapshot', 'generator'])['value'].unstack()
+    data = {'available': available, 'p_max': p_max, 'cost': cost, 'load': load}
+    coords = {
+        'generator': pd.Index(p_max.index, name='generator'),
+        'snapshot': pd.Index(load.index, name='snapshot'),
+    }
+    return data, coords
+
+
 # --------------------------------------------------------------------------
 # transport
 
@@ -220,6 +276,14 @@ def _ladder(sizes: dict[str, int], snapshots: Sequence[int], per_snapshot: int)
         write=_dispatch_data,
         eager_inputs=_dispatch_eager,
     ),
+    'sparse': Case(
+        name='sparse',
+        model=MODELS / 'sparse.yaml',
+        # nominal counts are the full product; SPARSE_DENSITY of it survives
+        ladder=_ladder({'generator': 100}, (100, 1_000, 10_000, 100_000), per_snapshot=100),
+        write=_sparse_data,
+        eager_inputs=_sparse_eager,
+    ),
     'transport': Case(
         name='transport',
         model=MODELS / 'transport.yaml',

bench/models/sparse.yaml

@@ -0,0 +1,47 @@
+# dispatch, but the mask actually bites.
+#
+# `dispatch.yaml` masks on `p_max > 0` and the generated p_max is always
+# positive, so its `where` removes nothing: it measures a fully dense coord
+# product, which is the eager lane's best case and the relational lane's
+# worst. Here `available` carries BOTH dims and is zero for most coordinates,
+# so the two representations of a mask are what is under test — row absence
+# relationally, NaN-padding in a dense array.
+#
+# Reading the leading dim is deliberate: it also puts the label frame on the
+# general (sorted) path rather than the factored one, so the two paths are
+# both covered by the ladder.
+dimensions:
+  snapshot:
+    dtype: int
+  generator:
+    dtype: str
+
+parameters:
+  available:
+    dims: [snapshot, generator]
+  p_max:
+    dims: [generator]
+  load:
+    dims: [snapshot]
+  cost:
+    dims: [generator]
+
+variables:
+  p:
+    foreach: [snapshot, generator]
+    where: "available > 0"
+    bounds:
+      lower: 0
+      upper: p_max
+
+constraints:
+  power_balance:
+    foreach: [snapshot]
+    equations:
+      - expression: sum(p, over=generator) == load
+
+objectives:
+  total_cost:
+    sense: minimize
+    equations:
+      - expression: p * cost

bench/profile_build.py

@@ -0,0 +1,121 @@
+"""Attribute build wall time to the SQL that spends it.
+
+``bench/run.py`` says *how much* slower we are; this says *where*. It wraps
+``DuckDBPyConnection.execute`` and tags every statement with the build step
+that issued it, so the output is a ranked list of statements rather than a
+single number.
+
+    uv run python -m bench.profile_build dispatch l
+    uv run python -m bench.profile_build transport m --memory-limit 4GB
+
+The wrapper adds Python overhead per call, so **absolute times here are not
+comparable to ``bench/run.py``** — there are only a few dozen statements, but
+the process is otherwise unoptimised. Read the shares, not the seconds; to
+quote a number, measure it with the harness.
+"""
+
+from __future__ import annotations
+
+import argparse
+import collections
+import time
+from pathlib import Path
+from typing import Any
+
+from bench import cases as bench_cases
+
+STEPS = (
+    '_create_param_table',
+    '_create_dim_tables',
+    '_build_variable',
+    '_build_constraint',
+    '_build_objective',
+)
+
+
+def _instrument(timings: dict[Any, list[float]], phase: dict[str, str]) -> None:
+    """Tag each executed statement with the build step that issued it."""
+    import duckdb
+
+    from farkas.relational.executor import DuckdbExecutor
+
+    original_execute = duckdb.DuckDBPyConnection.execute
+
+    def execute(self, sql, *args, **kwargs):
+        started = time.perf_counter()
+        try:
+            return original_execute(self, sql, *args, **kwargs)
+        finally:
+            elapsed = time.perf_counter() - started
+            key = (phase['now'], ' '.join(str(sql).split())[:88])
+            entry = timings.setdefault(key, [0.0, 0])
+            entry[0] += elapsed
+            entry[1] += 1
+
+    duckdb.DuckDBPyConnection.execute = execute
+
+    for name in STEPS:
+        original_step = getattr(DuckdbExecutor, name)
+
+        def wrap(step, label):
+            def wrapper(self, *args, **kwargs):
+                previous, phase['now'] = phase['now'], label
+                try:
+                    return step(self, *args, **kwargs)
+                finally:
+                    phase['now'] = previous
+
+            return wrapper
+
+        setattr(DuckdbExecutor, name, wrap(original_step, name))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('case', choices=sorted(bench_cases.CASES))
+    parser.add_argument('size', help='a rung of the case ladder, e.g. xs s m l')
+    parser.add_argument('--memory-limit', default='1GB')
+    parser.add_argument('--top', type=int, default=12, help='statements to list')
+    args = parser.parse_args()
+
+    timings: dict[Any, list[float]] = {}
+    phase = {'now': 'setup'}
+    _instrument(timings, phase)
+
+    import farkas as fk
+
+    case = bench_cases.CASES[args.case]
+    sources = case.data(case.shape(args.size))
+
+    import tempfile
+
+    with tempfile.TemporaryDirectory() as tmp:
+        started = time.perf_counter()
+        with fk.build(case.model, sources, memory_limit=args.memory_limit) as executor:
+            build = time.perf_counter() - started
+            phase['now'] = 'emit'
+            started = time.perf_counter()
+            executor.write_lp(Path(tmp) / 'model.lp')
+            emit = time.perf_counter() - started
+
+    print(f'\n{args.case}/{args.size} at {args.memory_limit}: build {build:.2f}s, emit {emit:.2f}s')
+    print('(instrumented — read the shares, not the seconds)\n')
+
+    by_step: dict[str, list[float]] = collections.defaultdict(lambda: [0.0, 0])
+    for (step, _), (elapsed, calls) in timings.items():
+        by_step[step][0] += elapsed
+        by_step[step][1] += calls
+    total = sum(v[0] for v in by_step.values()) or 1.0
+
+    print(f'{"step":24} {"seconds":>8} {"share":>7} {"calls":>7}')
+    for step, (elapsed, calls) in sorted(by_step.items(), key=lambda kv: -kv[1][0]):
+        print(f'{step:24} {elapsed:8.2f} {100 * elapsed / total:6.0f}% {calls:7d}')
+
+    print(f'\ntop {args.top} statements')
+    ranked = sorted(timings.items(), key=lambda kv: -kv[1][0])[: args.top]
+    for (step, sql), (elapsed, calls) in ranked:
+        print(f'  {elapsed:6.2f}s {100 * elapsed / total:4.0f}%  n={calls:<4d} [{step}]\n      {sql}')
+
+
+if __name__ == '__main__':
+    main()