exec, optimizer: TPC-H q21 parity (S5): join arena in 2^14-row blocks, 5/4 salted directory, unique-key chains, LatePayloadJoin

changelog.d/749.performance.md

@@ -0,0 +1,3 @@
+Hash joins keep their build side in fixed-size row blocks with a smaller, denser hash directory, and a join's
+unused columns are joined in only after the filtering joins above it: TPC-H q21 at scale factor 1000 runs in
+277 s instead of 474 s with no spill (laptop, cold run).

crates/krishiv-common/src/config/registry.rs

@@ -28,7 +28,8 @@ pub use lakehouse::{
     CATALOG_MAX_COLUMNS, CATALOG_MAX_METADATA_BYTES, CATALOG_REQUEST_TIMEOUT, DELTA_LOG_THREADS,
 };
 pub use optimizer::{
-    EXPERIMENTAL_GROUP_KEY_CODE_FORCE, GROUP_KEY_CODE, SHARE_REPEATED_SUBTREES, SUBQUERY_AS_WINDOW,
+    EXPERIMENTAL_GROUP_KEY_CODE_FORCE, GROUP_KEY_CODE, LATE_PAYLOAD_JOIN, SHARE_REPEATED_SUBTREES,
+    SUBQUERY_AS_WINDOW,
 };
 pub use result_cache::{RESULT_CACHE_ENABLED, RESULT_CACHE_MAX_BYTES, RESULT_CACHE_TTL};
 pub use server::{
@@ -97,6 +98,7 @@ pub static REGISTRY: &[&ConfigEntry] = &[
     &RESULT_CACHE_TTL,
     &RESULT_CACHE_MAX_BYTES,
     &SUBQUERY_AS_WINDOW,
+    &LATE_PAYLOAD_JOIN,
     &GROUP_KEY_CODE,
     &EXPERIMENTAL_GROUP_KEY_CODE_FORCE,
     &SHARE_REPEATED_SUBTREES,
@@ -798,7 +800,8 @@ pub static DETERMINISTIC_ORDER: ConfigEntry = ConfigEntry {
           seed, so an operator that spills splits its rows the same way on every run. Costs \
           parallelism where it falls back to one driver: meant for reproducing a result, \
           comparing two builds, and tests. Memory pressure from other queries in the process \
-          can still change when an operator spills, and so the order of what it returns. The \
+          can still change when an operator spills, or whether a hash join build copies its \
+          rows into blocks as they arrive, and so the order of what it returns. The \
           branches of a `UNION ALL` under an operator that must see all of them (a hash join \
           probing a union) take turns in branch order, so they lose their overlap too. Off by \
           default.",

crates/krishiv-common/src/config/registry/index.rs

@@ -170,6 +170,7 @@ pub static INDEX: &[(&str, &ConfigEntry)] = &[
         "krishiv.memory.reservationQuantum",
         &MEMORY_RESERVATION_QUANTUM,
     ),
+    ("krishiv.optimizer.latePayloadJoin", &LATE_PAYLOAD_JOIN),
     (
         "krishiv.python.worker.executables",
         &PYTHON_WORKER_EXECUTABLES,

crates/krishiv-common/src/config/registry/optimizer.rs

@@ -33,6 +33,23 @@ pub static SHARE_REPEATED_SUBTREES: ConfigEntry = ConfigEntry {
     spark_alias: None,
 };
 
+/// A dimension's payload joins above the semi and anti joins that never read it
+/// (`krishiv.optimizer.latePayloadJoin`).
+pub static LATE_PAYLOAD_JOIN: ConfigEntry = ConfigEntry {
+    key: "krishiv.optimizer.latePayloadJoin",
+    value_type: ValueType::Bool,
+    default: Some(ConfigValue::Bool(true)),
+    doc: "Join a dimension's payload columns (such as a name read only by the aggregate on top) \
+          after the semi, anti and outer joins that never read them, and leave a semi join on the \
+          dimension's keys where the join was, so those joins build and probe narrower rows. Applies \
+          when statistics show the payload is wide and the dimension is the smaller side. Results \
+          are the same either way.",
+    scope: Scope::Session,
+    settable_by: SettableBy::User,
+    version_added: "0.1.0",
+    spark_alias: None,
+};
+
 /// Whether an aggregate's group keys from one dimension travel as one integer code (P2b D2).
 pub static GROUP_KEY_CODE: ConfigEntry = ConfigEntry {
     key: "krishiv.optimizer.groupKeyCode",

crates/krishiv-common/src/config/tests.rs

@@ -1329,6 +1329,21 @@ fn fuse_probe_select_defaults_on_and_parses() {
     );
 }
 
+#[test]
+fn late_payload_join_defaults_on_and_parses() {
+    let mut c = Config::new();
+    assert_eq!(
+        c.get(&LATE_PAYLOAD_JOIN).and_then(ConfigValue::as_bool),
+        Some(true)
+    );
+    c.set("krishiv.optimizer.latePayloadJoin", "false").unwrap();
+    assert_eq!(
+        c.get(&LATE_PAYLOAD_JOIN).and_then(ConfigValue::as_bool),
+        Some(false)
+    );
+    assert!(c.set("krishiv.optimizer.latePayloadJoin", "maybe").is_err());
+}
+
 #[test]
 fn python_worker_entries_parse_and_default() {
     let mut c = Config::new();

crates/krishiv-exec/CLAUDE.md

@@ -36,7 +36,8 @@ idle after `krishiv.exec.join.idleSpillAfter`, so the arbitrator asks them befor
 30. Each probe driver then holds the build's level-0 tables, `build.rs` `Level0`, whose memory includes the per-partition
 reservations moved at the seal; the last probe driver done after the flush barrier releases them, so a downstream
 sink's final phase gets that memory; a `Prebuilt` broadcast build is never released) · `grace.rs` spill partitions,
-flush-phase grace join · `broadcast.rs`
+flush-phase grace join · `block.rs` a
+partition's chunks kept few: the probe side's tiered merge, the build side's cut into the arena's 2^14-row blocks · `broadcast.rs`
 (P2-T177 unit 3) `BroadcastTables`, a worker resource (`WorkerContext::with_broadcasts`): a broadcast join over a cut
 broadcast exchange builds its table once per job and worker (its own query over the resource's pool, no spill, its
 reservations never abort a query) and every task probes it through a `PrebuiltSink` that republishes its runtime