Improve planning and continue cleaning

.clang-tidy

@@ -1,3 +1,9 @@
+# misc-use-internal-linkage is disabled below: the runtime's foo.cpp +
+# foo_internal.hpp split shares helpers across a few TUs through named *_detail
+# namespaces. Analyzing one TU the check can't see the cross-TU use and tells us
+# to make those `static` (which breaks the link); the genuinely TU-local helpers
+# interleaved in the same blocks would each need call-site requalification to
+# move to an anonymous namespace. Style only (inlining / symbol-table size).
 Checks: >
   -*,
   bugprone-*,
@@ -26,7 +32,8 @@ Checks: >
   -cppcoreguidelines-owning-memory,
   -bugprone-easily-swappable-parameters,
   -cppcoreguidelines-pro-type-union-access,
-  -cppcoreguidelines-pro-type-vararg
+  -cppcoreguidelines-pro-type-vararg,
+  -misc-use-internal-linkage
 CheckOptions:
   # libstdc++ implementation headers are not portable include providers.
   # Prefer the public standard-library header for every standard declaration.

README.md

@@ -266,7 +266,8 @@ trades[update { (delta, gamma) = compute_greeks(price) }]
 
 User-defined functions can already require a minimum input table schema with
 `DataFrame<{ ... }>` parameter types. Declared columns must exist with the
-right types; extra columns are allowed:
+right types; extra columns are allowed and pass through, but cannot be named in
+the function body unless the parameter schema declares them:
 
 ```
 fn top_two_salaries(df: DataFrame<{ salary: Int64 }>) -> DataFrame effects {} {
@@ -845,6 +846,7 @@ directory.
 :schema <table>          Show column names and types
 :head <table> [n]        Show first n rows (default 10)
 :peek <expr>             Evaluate and compactly display an expression
+:explain <expr>          Show the physical-plan capability without executing it
 :describe <table> [n]    Schema + first n rows
 :doc <name>              Show docs/signature for a binding or built-in
 ?name                    Shorthand for :doc <name>

SPEC.md

@@ -3152,6 +3152,11 @@ than an exact-match type.
 - `DataFrame<{ departmentId: Int, salary: Int }>` requires both columns; other
   columns remain permitted.
 
+Only the declared fields are statically nameable inside the function. Permitted
+extra columns pass through unchanged, but the function cannot reference one by
+name unless it adds that column to its parameter schema. The open remainder is
+a validation allowance, not a source of dynamically discoverable column names.
+
 This contract is checked at **call time**. A missing required column, or a
 required column with the wrong type, is a call-time error that names the
 parameter and the offending column. The same contract applies to a function's

include/ibex/ir/join_output.hpp

@@ -48,6 +48,33 @@ struct JoinOutputColumn {
     auto operator==(const JoinOutputColumn&) const -> bool = default;
 };
 
+/// One equijoin key resolved against the ordered physical columns of both
+/// inputs. Names are a logical/schema concern; build and probe kernels consume
+/// these stable positions.
+struct JoinKeyColumns {
+    std::size_t left_index = 0;
+    std::size_t right_index = 0;
+
+    auto operator==(const JoinKeyColumns&) const -> bool = default;
+};
+
+/// The complete column-name resolution for one join: positional key bindings
+/// plus the authoritative output gather/rename plan. This is the join analogue
+/// of `ColumnNameMap`: consumers resolve names once, then share this value
+/// instead of independently looking them up or reconstructing output names.
+struct JoinColumnMapping {
+    /// Ordered physical layouts against which every position below was
+    /// resolved. A lazy child can consume predicate-only columns before the
+    /// join boundary; execution detects that change once and re-resolves the
+    /// complete mapping against the concrete tables.
+    std::vector<std::string> left_input_names;
+    std::vector<std::string> right_input_names;
+    std::vector<JoinKeyColumns> keys;
+    std::vector<JoinOutputColumn> output;
+
+    auto operator==(const JoinColumnMapping&) const -> bool = default;
+};
+
 /// The single authority on a join's output column list and naming.
 ///
 /// IR schema inference, the materialized interpreter, the chunked executor and
@@ -75,4 +102,14 @@ struct JoinOutputColumn {
                                     const JoinSuffixPolicy& suffix = {})
     -> std::expected<std::vector<JoinOutputColumn>, std::string>;
 
+/// Resolve every textual join key to its input position and compute the output
+/// plan through `plan_join_output`. Known schemas call this during physical
+/// planning; lazy/unknown schemas call it once when their concrete columns
+/// first reach the build/probe barrier.
+[[nodiscard]] auto resolve_join_columns(JoinKind kind, const std::vector<JoinKey>& keys,
+                                        std::span<const std::string_view> left_names,
+                                        std::span<const std::string_view> right_names,
+                                        const JoinSuffixPolicy& suffix = {})
+    -> std::expected<JoinColumnMapping, std::string>;
+
 }  // namespace ibex::ir

include/ibex/runtime/interpreter.hpp

@@ -703,6 +703,18 @@ struct ExecutionContext {
     /// chunked join, and the lazy-table keep-rows scan) and must agree.
     bool parallel_join_probe = true;
 
+    /// Child subtrees the materialized-call fallback already built through the
+    /// physical path (`build_materialized_fallback`), keyed by node pointer.
+    /// Null for every ordinary call. When set, `interpret_node` returns the
+    /// pre-built table for a node listed here instead of recursing into it, so a
+    /// filtered or projected input to a fallback breaker keeps the fused
+    /// parallel scan `build_operator` gave it rather than being re-evaluated
+    /// whole-table and serial. Only the fallback node's direct children appear;
+    /// deeper descendants are interpreted normally. Not owned — the tables live
+    /// in the fallback frame for the duration of the `interpret_node` call.
+    const std::vector<std::pair<const ir::Node*, const Table*>>* pre_materialized_children =
+        nullptr;
+
     /// Look up a deferred scan by its plan (instance) name, or null if there is
     /// no registry or no matching entry.
     [[nodiscard]] auto deferred_scan(const std::string& name) const -> const DeferredScan* {

libs/data_gen/data_gen.ibex

@@ -17,21 +17,30 @@ extern fn gen_ticks(
     volatility: Float64 = 0.5,
     interval_ms: Float64 = 1000.0,
     start_ts_ms: Int = 0
-) -> DataFrame from "data_gen.hpp";
+) -> DataFrame<{
+    timestamp: Timestamp,
+    symbol: String,
+    price: Float64,
+    volume: Int64
+}> from "data_gen.hpp";
 
 // Gaussian random walk: value[0] = start, value[i] = value[i-1] + N(0, step_std).
-extern fn gen_walk(n: Int, start: Float64 = 0.0, step_std: Float64 = 1.0) -> DataFrame from "data_gen.hpp";
+extern fn gen_walk(n: Int, start: Float64 = 0.0, step_std: Float64 = 1.0)
+    -> DataFrame<{ value: Float64 }> from "data_gen.hpp";
 
 // n iid samples from N(mean, stddev) in a single "value" column.
-extern fn gen_normal(n: Int, mean: Float64 = 0.0, stddev: Float64 = 1.0) -> DataFrame from "data_gen.hpp";
+extern fn gen_normal(n: Int, mean: Float64 = 0.0, stddev: Float64 = 1.0)
+    -> DataFrame<{ value: Float64 }> from "data_gen.hpp";
 
 // n iid samples from Uniform[low, high) in a single "value" column.
-extern fn gen_uniform(n: Int, low: Float64 = 0.0, high: Float64 = 1.0) -> DataFrame from "data_gen.hpp";
+extern fn gen_uniform(n: Int, low: Float64 = 0.0, high: Float64 = 1.0)
+    -> DataFrame<{ value: Float64 }> from "data_gen.hpp";
 
 // n sequential string ids "<prefix>0" .. "<prefix>(n-1)".
 extern fn gen_ids(n: Int, prefix: String = "row") -> DataFrame from "data_gen.hpp";
 
 // Static reference data: one row per distinct symbol, with columns
 // symbol, name, sector, currency, lot_size, tick_size. Deterministic (no RNG),
 // so it joins against `gen_ticks` output on `symbol`.
-extern fn gen_reference(symbols: String = "AAPL,MSFT,GOOG") -> DataFrame from "data_gen.hpp";
+extern fn gen_ids(n: Int, prefix: String = "row")
+    -> DataFrame<{ id: String }> from "data_gen.hpp";

plans/README.md

@@ -1,7 +1,8 @@
 # Plans Index
 
-Status of every plan in this directory, grouped by lifecycle. Statuses
-re-verified against the source tree on **2026-08-27** (that pass also compacted
+Status of every plan in this directory, grouped by lifecycle. Statuses were
+initially re-verified against the source tree on **2026-08-27** (the
+kernel-pipeline entry was re-verified again on 2026-08-29); that pass compacted
 the four largest plans — `parallelism-overview`, `query-shape-conformance`,
 `runtime-multithreading`, `kernel-pipeline-execution` — plus
 `owned-agg-per-chunk-barrier`, moving their measurement diaries to git history
@@ -15,13 +16,13 @@ history).
 | Plan | Status | What's actually left |
 |---|---|---|
 | [beat-polars-plan.md](beat-polars-plan.md) | **Umbrella plan** for the multi-core push (proposed; §8 keeps the baseline record and dead ends). 2026-08-27 update corrected two stale q10 diagnoses (carried group fields optimized during discovery, ~−10.5%; the "36ms serial join build" was inclusive attribution). | Target: implied parallel fraction 44% → 60–65%. Workstreams W1 parallel inner join, W2 aggregate residue, W3 scheduler slice 2, W4 chunked `let` bindings, W5 small-query-tax guard. Points into pipelined-execution + runtime-multithreading + kernel-pipeline for mechanism. |
-| [kernel-pipeline-execution-plan.md](kernel-pipeline-execution-plan.md) | **Phase 2 complete** except `KernelContext` (deliberately unbuilt); **Phase 3 started** (executor-owned ordered handoff, islands dissolved into a pipeline mode); **Phase 4 construction ownership done** (backlog 116→6 breakers, plan describes 97% of real-work nodes) but **decomposition not started** — operators are unchanged, so the exit criterion is not met. Architectural successor: typed logical IR, physical pipelines, morsel executor, templated kernel library; not a JIT. | Next: split the join operator into scheduled `HashBuild`+`HashProbe` (join build-side split DONE 2026-08-25, q21 −8.5%; probe-as-pipeline-step BUILT off-by-default, fires on 1/22 queries); aggregate phase decomposition (blocked-first on an already-broken determinism constraint — `try_owned_pair` vs the serial path disagree bit-for-bit at ≥65536 rows, no test); port `Tail`/`TopK`/`FilterHead`/`FilterTail`; split `chunked.cpp` by ownership. |
+| [kernel-pipeline-execution-plan.md](kernel-pipeline-execution-plan.md) | **Phase 2 complete** except `KernelContext` (deliberately unbuilt); Phase 3 handoff/island/raw-thread work complete, with accounting and DOP/memory budgets deferred; Phase 4 construction ownership **and fan-out authority** done (backlog 116→6 breakers, plan describes 97% of real-work nodes). Streaming inner joins have typed `HashBuild`/`HashProbe` nodes and positional `JoinColumnMapping`. Streaming aggregates have positional `AggregateColumnMapping`, authoritative partition/finalize policy, and a typed Discovery → Accumulation → FinalOrdering → Emission hash-fallback chain. The serial coordinator invokes all four nodes through a bounded discovery transfer or explicit fused marker, with independent profile rows. Executor-seam mutations prove mappings, policies, and structural edges are consumed or rejected. Known closed schemas bind during planning; lazy/open schemas bind once at execution. Semi/anti retains its separate streaming operator. Architectural successor: typed logical IR, physical pipelines, morsel executor, templated kernel library; not a JIT. | Next: attach aggregate fan-out policy to each structural node, admit it phase by phase, then split `chunked.cpp` by ownership. |
 | [benchmark-perf-priorities.md](benchmark-perf-priorities.md) | Living reference | P0–P2 resolved/landed; rolling min/max optimized. Open: suite trimming (pin sqlite + data.table frollapply cells, duckdb at 3 scales); P4 `tanh` deferred pending accuracy-vs-speed call; P3 ohlc scatter-bound (negative result recorded — don't re-attempt naive fusion); re-check rolling_mean on AWS after the July 2026 regression fix |
 | [benchmark-coverage-plan.md](benchmark-coverage-plan.md) | ~95% done | #9 ClickHouse EWMA (needs arrayFold workaround); #10 DataFusion `fill_forward/backward` + `tf_asof_join` |
 | [count-window-plan.md](count-window-plan.md) | Implemented (interpreter + codegen) | Per-call count/duration windows work (`__window_n`/`__window_ns` in lower.cpp + window.cpp), and the compiled path (`ibex_compile`) is at parity. Open: `window N rows` block syntax and tuple-field `update` inside `window` (interpreter doesn't support that combo either, so codegen correctly still rejects it). The old monotonic-deque follow-up for `rolling_min`/`rolling_max` is done. |
 | [non-row-local-filter-plan.md](non-row-local-filter-plan.md) | Stage 1 shipped | `lag`/`lead`/`is_null` in filter work. Remaining: `rank(...)` in filter/select with `by`, explicit `order {}` context, rolling functions in filter (`price > rolling_mean(price)`) |
 | [bigger-than-ram-plan.md](bigger-than-ram-plan.md) | Phase 4 bullet 1 of 4 done | Out-of-core execution. Done: chunked/streaming `read_parquet` (branch `chunked-parquet-read`; ~6.5× lower peak RSS, ~1.7× faster, verified local + AWS). Next: column projection pushdown, row-group stats pushdown, directory/Hive datasets (rest of Phase 4), then Phase 1 spill infrastructure (prerequisite for Phases 2–3, 6–7: external sort, out-of-core join, adaptive spill selection) |
-| [runtime-multithreading-plan.md](runtime-multithreading-plan.md) | Phase 1 landed and **ON by default** (parallel islands = two-phase filter + metadata-tail, every shape beats serial); Phase 3a complete; Phase 3b first source slice landed; Phase 4 items 1–2 landed, item 3 RETIRED (the join gap was Categorical probe keys hashed as *text*, not threading), item 4 part-done. **Nomenclature: `IBEX_THREADS` → `IBEX_CORES`; `IBEX_PARALLEL` removed (serial is `IBEX_CORES=1`).** | The PDS-H multithreading gap is **Phase 4 (parallel barriers)**, not sources — only 5/22 queries even form an island. Next: group-by string/int/generic hash paths + `distinct`; the LazyTable Synchronization Contract (written, unimplemented — Phase 3b's foundation); Phase 2 deterministic RNG (designed, not started). Re-measure at a larger scale before ranking — the threading share of a gap grows with row count. |
+| [runtime-multithreading-plan.md](runtime-multithreading-plan.md) | Row-local morsel-parallel pipelines are **ON by default**; Phase 3a is complete; Phase 3b's first source slice landed; Phase 4 items 1–2 landed, item 3 RETIRED (the join gap was Categorical probe keys hashed as *text*, not threading), item 4 part-done. **Nomenclature: `IBEX_THREADS` → `IBEX_CORES`; `IBEX_PARALLEL` removed (serial is `IBEX_CORES=1`).** | The PDS-H multithreading gap is **parallel barriers**, not sources. Next: group-by string/int/generic hash paths + `distinct`; the LazyTable Synchronization Contract (written, unimplemented — Phase 3b's foundation); Phase 2 deterministic RNG (designed, not started). Re-measure at a larger scale before ranking — the threading share of a gap grows with row count. |
 | [join-perf-plan.md](join-perf-plan.md) | Items 1–3 done (2026-07-14; q09 −23%, q13 −30%) | Join/group-by performance findings; see the file's Results section. beat-polars points here for join mechanism |
 | [owned-agg-per-chunk-barrier-plan.md](owned-agg-per-chunk-barrier-plan.md) | High-cardinality partition-owned aggregation vs Polars streaming. q18 (single-Int64 `Sum`) largely closed by the async hot/cold rewrite (**−33%**); the parallel finalize merge landed for all owned paths (**−7%**); q21's ordered-run `Count` finalize + emit fusion landed (**−11.2% SF-4**). | q20/`PairIntKey` (the hot table doesn't help a scattered composite key — a per-partition `CardinalitySketch` is the candidate); q21's remaining wall is the per-chunk accumulate orchestration + a duplicate lineitem decode + a 40ms serial hash-join build. **Do NOT touch `part_count` or serial-`reserve` the maps** — both measured dead ends. |
 | [grouped-chunkview-update-plan.md](grouped-chunkview-update-plan.md) | Mostly complete — `update …, by k` runs off an immutable `GroupedRowPlan` (CSR) instead of gather → per-group `Table` → scatter. Sub-plan of kernel-pipeline Phase 2. | Remaining materialized shapes: `rank`, variable-width ordered state, `window`-clause `lag`/`lead`. |
@@ -40,7 +41,6 @@ history).
 
 | Plan | Notes |
 |---|---|
-| [pipelined-execution-plan.md](pipelined-execution-plan.md) | **Phase 2 partially implemented.** Multi-chunk correctness, lazy row-group streaming, concurrent scan decoding, ordered source→map overlap, and a bounded streamed join-probe handoff landed. SF-1 improves 2.5–4.6% at 2–8 cores; SF-4 improves 0.6–3.7% at 4–8. The two-core SF-4 crossover (+6.9%) and q18 are explicit follow-ups. Open: progress-aware admission/backpressure and general breaker scheduling |
 | [parallel-chunkview-output-plan.md](parallel-chunkview-output-plan.md) | ChunkView kernel output protocol under parallel execution — `update_table` stays the semantic/ownership authority, a direct kernel plans a field and writes one assigned range. Sub-plan of kernel-pipeline Phase 2 (see the "Parallel chunk updates" entries). |
 | [radix-partitioned-groupby.md](radix-partitioned-groupby.md) | Noted, not built. High-cardinality group-by is memory-bound; radix partitioning remains a q18/q20 mechanism. Q10 no longer reaches the generic mixed-key ceiling (2026-08-27: FD reduction + discovery-time `First` gathering handle that shape). |
 | [exists-subquery-plan.md](exists-subquery-plan.md) | Proposal: `exists(table_expr)` as a boolean subquery term — semi/anti/mark joins and the residual-predicate case |

plans/beat-polars-plan.md

@@ -4,11 +4,11 @@ Status: **proposed.** Written 2026-08-16 on branch `streaming-scan-units`, after
 the pipelined-execution Phase 2 slices (streamed scan units, concurrent decode,
 source→map pipeline, join-probe handoff) landed. This is the umbrella plan: it
 sets the target, decomposes the remaining gap into workstreams with expected
-payoffs, and sequences them. It supersedes nothing — it points into
-`pipelined-execution-plan.md`, `runtime-multithreading-plan.md`, and
-`join-perf-plan.md` for mechanism (the former fourth input,
-`chunked-execution-plan.md`, was removed from the tree 2026-08-22; its
-open items live in the kernel-pipeline plan and bigger-than-ram).
+payoffs, and sequences them. It points into
+`kernel-pipeline-execution-plan.md`, `runtime-multithreading-plan.md`, and
+`join-perf-plan.md` for mechanism. The former pipelined- and chunked-execution
+plans were removed as superseded; their remaining work lives in the kernel
+pipeline plan and bigger-than-ram plan.
 Absorbed `pds.md` (the 2026-08-11 status snapshot this grew out of) on
 2026-08-22: §5 and §6 took its still-unique traps and dead ends, §8 keeps
 its baseline record.
@@ -193,8 +193,8 @@ is, so every workstream is stated as "serial ms attacked".
 ## 2. Where the serial time is
 
 Calling-thread self-ms summed over the suite at 8 cores, with worker help drawn
-(`pool_work / self`), from `pipelined-execution-plan.md` ("Where the serial
-time actually is", post-semi-join numbers):
+(`pool_work / self`), from the archived pipelined-execution measurement study
+(git history; post-semi-join numbers):
 
 | operator | self ms | pool ms | worker help | verdict |
 |---|---:|---:|---:|---|
@@ -512,7 +512,7 @@ W3.1 progress-aware admission       ── next scheduler slice; fixes 2-core cl
 W3.2 q18/q22 rundown                ── decides W4's priority
 W4 chunked let bindings             ── design note first; largest structural risk
 W3.3+ aggregate-output stages,
-      general scheduler, island retirement
+      explicit breaker phases and general scheduler
 W2.3 parallel gid                   ── only if still dominant after the above
 ```