diff --git a/crates/krishiv-exec/benches/hash_join.rs b/crates/krishiv-exec/benches/hash_join.rs index 7a806452a..64f76f421 100644 --- a/crates/krishiv-exec/benches/hash_join.rs +++ b/crates/krishiv-exec/benches/hash_join.rs @@ -78,6 +78,7 @@ fn plan_with( build_keys: vec![0], null_safe: vec![false], filter: None, + single_match: false, spill, }; let scan = diff --git a/crates/krishiv-exec/examples/join_index_matrix.rs b/crates/krishiv-exec/examples/join_index_matrix.rs index 73537b6fc..ae8aa008f 100644 --- a/crates/krishiv-exec/examples/join_index_matrix.rs +++ b/crates/krishiv-exec/examples/join_index_matrix.rs @@ -359,6 +359,7 @@ fn run_join( build_keys: vec![0], null_safe: vec![false], filter: None, + single_match: false, spill: None, }; let scan = |b: &[Batch]| -> Arc { diff --git a/crates/krishiv-exec/src/join/grace.rs b/crates/krishiv-exec/src/join/grace.rs index fc2376c1b..7606aeab1 100644 --- a/crates/krishiv-exec/src/join/grace.rs +++ b/crates/krishiv-exec/src/join/grace.rs @@ -1642,6 +1642,7 @@ mod tests { build_keys: vec![0], null_safe: vec![false], filter: None, + single_match: false, spill: Some(scope.clone()), }) .unwrap(); diff --git a/crates/krishiv-exec/src/join/mod.rs b/crates/krishiv-exec/src/join/mod.rs index 79e3396a0..f6f8d3ef5 100644 --- a/crates/krishiv-exec/src/join/mod.rs +++ b/crates/krishiv-exec/src/join/mod.rs @@ -1,7 +1,9 @@ // SPDX-License-Identifier: Apache-2.0 //! Hash join (P2-T052a; ADR 0011, E02): inner, left outer, left semi, left -//! anti and null-aware anti (`NOT IN`) equi-joins with an optional residual -//! filter and partitioned grace-hash spill. +//! anti, null-aware anti (`NOT IN`), existence (Spark's `ExistenceJoin`: +//! `EXISTS` and `IN` outside a filter conjunct) and single-match left outer +//! (scalar subquery) equi-joins with an optional residual filter and +//! partitioned grace-hash spill. //! //! # Shape //! The **build** side is the right input: a [`Sink`] ending a child @@ -9,7 +11,11 @@ //! [`Operator`] in the parent pipeline that reads the finalized build //! through [`TaskContext::dependency`](crate::TaskContext::dependency). //! Output columns are the probe columns, then (inner and left outer) every -//! build column; build columns are nullable in a left outer join. +//! build column; build columns are nullable in a left outer join. An +//! existence join outputs the probe columns plus one non-nullable boolean +//! column, `exists`. A +//! single-match left outer join outputs what a left outer join does, and +//! fails when a probe row matches more than once. //! [`HashJoinNode`] wires both. //! //! # Spark semantics @@ -26,6 +32,22 @@ //! keeps every probe row (NULL keys too); a NULL key on the build side //! keeps none; otherwise a probe row is kept if its key is not NULL and //! matches no build key. +//! - Existence joins emit every probe row once plus `exists`: TRUE when +//! some build row matches it (keys equal and the residual filter TRUE), +//! else FALSE. `exists` is never NULL, as in Spark 4.2 (checked on Spark +//! 4.2.0 in every context that plans an `ExistenceJoin`: `k IN (subquery)` +//! is FALSE, not NULL, when the subquery holds a NULL or `k` is NULL). +//! `IN` is the existence join on `k = subquery column`; `NOT IN` is the +//! negation of the existence join whose residual is +//! `(k = c) OR isnull(k = c)` (with the correlation equalities as keys, or +//! by nested loops when there are none), exactly the conditions Spark +//! plans. A probe row's `exists` depends only on the build rows with its +//! key, so the join needs no build-wide facts and runs broadcast or +//! partitioned alike. +//! - A single-match left outer join (a correlated scalar subquery without +//! an aggregate) emits what a left outer join does, and fails the query +//! with `SCALAR_SUBQUERY_TOO_MANY_ROWS` when a probe row matches more +//! than one build row. //! - The residual filter sees probe columns then build columns of each //! candidate pair; NULL is not a match. //! @@ -168,6 +190,11 @@ pub enum JoinType { LeftAnti, /// `NOT IN`: see the module docs. NullAwareAnti, + /// Spark's `ExistenceJoin` (`EXISTS`/`IN` outside a filter conjunct): + /// every probe row once, plus a non-nullable boolean column `exists`, + /// TRUE when some build row matches (keys equal and the residual filter + /// TRUE), else FALSE. See the module docs. + Existence, /// Build rows with at least one match, each once: a left semi join /// that builds its left input. Planned data only for now: no operator /// runs it yet. @@ -212,6 +239,10 @@ pub struct JoinSpec { /// The residual condition, over the probe columns then the build /// columns of each candidate pair; NULL does not match. pub filter: Option>, + /// A left outer join that fails with `SCALAR_SUBQUERY_TOO_MANY_ROWS` + /// when a probe row matches more than one build row (a correlated + /// scalar subquery without an aggregate). Only `LeftOuter` joins set it. + pub single_match: bool, /// Where to spill; `None` never spills (memory pressure then fails the /// query through the arbitrator). pub spill: Option, @@ -355,6 +386,9 @@ impl Counters { /// State shared by a join's build sink and probe operator. pub(crate) struct Shared { join_type: JoinType, + /// A left outer join that fails when a probe row matches more than + /// once (see [`JoinSpec::single_match`]). + single_match: bool, probe_keys: Vec, build_keys: Vec, keys: keys::KeyCodec, @@ -405,6 +439,14 @@ fn unsupported(feature: &str) -> KrishivError { .into() } +/// The probe columns plus Spark's non-nullable `exists` column: the output +/// of an existence join (hash or nested loop). +pub(crate) fn existence_schema(probe: &Schema) -> SchemaRef { + let mut fields: Vec = probe.fields().iter().map(|f| f.as_ref().clone()).collect(); + fields.push(Field::new("exists", DataType::Boolean, false)); + Arc::new(Schema::new(fields)) +} + impl Shared { fn new(spec: JoinSpec) -> Result { let JoinSpec { @@ -415,6 +457,7 @@ impl Shared { build_keys, null_safe, filter, + single_match, spill, } = spec; let n = probe_keys.len(); @@ -423,6 +466,11 @@ impl Shared { "a hash join needs matching, non-empty key lists".to_owned(), )); } + if single_match && join_type != JoinType::LeftOuter { + return Err(internal( + "single match on a join that is not left outer".to_owned(), + )); + } if join_type.builds_preserved() { return Err(unsupported( "hash joins that return rows of their build side by their matches", @@ -491,6 +539,7 @@ impl Shared { let schema = match join_type { JoinType::Inner => joined(false), JoinType::LeftOuter => joined(true), + JoinType::Existence => existence_schema(&probe_schema), _ => Arc::clone(&probe_schema), }; // Build columns as payload rows: only when every build column is @@ -526,6 +575,7 @@ impl Shared { }; Ok(Self { join_type, + single_match, probe_keys, build_keys, arrow_arena: table::arena_schema(arrow_format)?, @@ -670,9 +720,11 @@ impl HashJoin { /// /// # Errors /// `INTERNAL_ERROR` for keys that are missing, out of range or of - /// different types on the two sides; `KRISHIV_UNSUPPORTED_FEATURE` for a + /// different types on the two sides, and for `single_match` on a join + /// that is not left outer; `KRISHIV_UNSUPPORTED_FEATURE` for a /// null-aware anti join over several keys, null-safe keys or with a - /// residual filter (Spark plans those as a nested-loop join). + /// residual filter (Spark plans those as a nested-loop join), and for + /// the joins that return rows of their build side by their matches. pub fn try_new(spec: JoinSpec) -> Result { Ok(Self(Arc::new(Shared::new(spec)?))) } @@ -780,6 +832,7 @@ mod tests { build_keys: (0..keys).collect(), null_safe: vec![false; keys], filter: filter.then(|| Arc::new(crate::FnPredicate(all)) as Arc), + single_match: false, spill: None, } } @@ -813,6 +866,7 @@ mod tests { JoinType::LeftSemi, JoinType::LeftAnti, JoinType::NullAwareAnti, + JoinType::Existence, ] { assert!(!t.builds_preserved(), "{t:?}"); } @@ -846,6 +900,35 @@ mod tests { HashJoin::try_new(spec(JoinType::LeftSemi, 1, false)).unwrap_or_else(|e| panic!("{e}")); assert_eq!(j.schema().fields().len(), 2); assert!(format!("{j:?}").contains("LeftSemi")); + // Existence joins carry the probe columns plus Spark's + // non-nullable `exists`, over any number of keys, null-safe or + // not, with or without a residual condition. + for (keys, filter) in [(1, false), (2, true)] { + let mut s = spec(JoinType::Existence, keys, filter); + s.null_safe = vec![true; keys]; + let j = HashJoin::try_new(s).unwrap_or_else(|e| panic!("{e}")); + assert_eq!(j.schema().fields().len(), 3); + assert_eq!( + j.schema().field(2), + &Field::new("exists", DataType::Boolean, false) + ); + assert!(format!("{j:?}").contains("Existence")); + } + // Single match is a flag on the left outer join only. + let mut s = spec(JoinType::LeftOuter, 1, false); + s.single_match = true; + assert!(HashJoin::try_new(s).is_ok()); + for t in [ + JoinType::Inner, + JoinType::LeftSemi, + JoinType::LeftAnti, + JoinType::NullAwareAnti, + JoinType::Existence, + ] { + let mut s = spec(t, 1, false); + s.single_match = true; + assert!(HashJoin::try_new(s).is_err(), "{t:?}"); + } } /// Rows taken from a batch of view columns keep only their own bytes diff --git a/crates/krishiv-exec/src/join/nested.rs b/crates/krishiv-exec/src/join/nested.rs index 3f2052b7e..441d6a587 100644 --- a/crates/krishiv-exec/src/join/nested.rs +++ b/crates/krishiv-exec/src/join/nested.rs @@ -11,7 +11,11 @@ //! - inner: the matching pairs; //! - left outer: the matching pairs, and each probe row without a match //! with NULL build columns once the chunk met every build row; -//! - left semi / anti: each probe row with / without a match, once. +//! - left semi / anti: each probe row with / without a match, once; +//! - existence: each probe row once with `exists` (whether it matched; +//! never NULL). Spark plans `k NOT IN (uncorrelated subquery)` outside a +//! filter conjunct this way, as the negation of the existence join on +//! `(k = c) OR isnull(k = c)`. //! //! Rows must be insert-only (weight 1); weighted input is //! `KRISHIV_UNSUPPORTED_FEATURE` until IVM, as for the hash join's build. @@ -39,7 +43,7 @@ pub(crate) const TARGET_PAIRS: usize = 8192; /// What a nested-loop join joins. #[derive(Debug)] pub struct NestedLoopSpec { - /// Inner, left outer, left semi or left anti. + /// Inner, left outer, left semi, left anti or existence. pub join_type: JoinType, /// The probe (left) input's schema. pub probe_schema: SchemaRef, @@ -94,6 +98,7 @@ impl NestedLoopJoin { let schema = match spec.join_type { JoinType::Inner => Arc::clone(&pairs), JoinType::LeftOuter => Arc::new(Schema::new(fields(true))), + JoinType::Existence => super::existence_schema(&spec.probe_schema), _ => Arc::clone(&spec.probe_schema), }; Ok(Self(Arc::new(Shared { @@ -377,8 +382,9 @@ impl ProbeOp { .map_err(|e| internal(e.to_string())) } - /// Once a chunk met every build row: semi and anti rows, and the - /// null-padded probe rows of a left outer join. + /// Once a chunk met every build row: semi and anti rows, the + /// null-padded probe rows of a left outer join, and every probe row with + /// its `exists` of an existence join. fn unmatched( &self, probe: &RecordBatch, @@ -391,6 +397,7 @@ impl ProbeOp { .filter(|(_, m)| match self.0.spec.join_type { JoinType::LeftSemi => **m, JoinType::LeftAnti | JoinType::LeftOuter => !**m, + JoinType::Existence => true, _ => false, }) .map(|(i, _)| { @@ -409,10 +416,16 @@ impl ProbeOp { .map_err(|e| internal(e.to_string())) }) .collect::>>()?; - if self.0.spec.join_type == JoinType::LeftOuter { - for f in self.0.spec.build_schema.fields() { - columns.push(new_null_array(f.data_type(), idx.len())); + match self.0.spec.join_type { + JoinType::LeftOuter => { + for f in self.0.spec.build_schema.fields() { + columns.push(new_null_array(f.data_type(), idx.len())); + } + } + JoinType::Existence => { + columns.push(Arc::new(BooleanArray::from(matched.to_vec())) as ArrayRef); } + _ => {} } let options = RecordBatchOptions::new().with_row_count(Some(idx.len())); RecordBatch::try_new_with_options(Arc::clone(&self.0.schema), columns, &options) diff --git a/crates/krishiv-exec/src/join/probe.rs b/crates/krishiv-exec/src/join/probe.rs index 8dca8d045..7678d1cbc 100644 --- a/crates/krishiv-exec/src/join/probe.rs +++ b/crates/krishiv-exec/src/join/probe.rs @@ -6,9 +6,9 @@ use std::sync::Arc; -use arrow_array::{Array, ArrayRef, Int64Array, UInt32Array, new_null_array}; +use arrow_array::{Array, ArrayRef, BooleanArray, Int64Array, UInt32Array, new_null_array}; use futures::future::BoxFuture; -use krishiv_common::error::{ArithmeticOverflow, KrishivError}; +use krishiv_common::error::{ArithmeticOverflow, KrishivError, ScalarSubqueryTooManyRows}; use krishiv_memory::SpillSignal; use krishiv_types::batch::{Batch, Weights}; @@ -230,6 +230,14 @@ impl Cursor { }; for (j, &i) in probe.iter().enumerate() { if mask.as_ref().is_none_or(|k| k.value(j)) { + // A single-match join holds at most one match per probe + // row: the pairs of one row arrive in order, so a row that + // is already marked matched this row twice. The mark lives + // in `state`, so it survives across output batches, the + // in-memory and grace phases alike. + if shared.single_match && self.state[i as usize] == MATCHED { + return Err(too_many_rows()); + } self.state[i as usize] = MATCHED; } } @@ -274,6 +282,8 @@ impl Cursor { } JoinType::LeftSemi => MATCHED, JoinType::LeftOuter | JoinType::LeftAnti | JoinType::NullAwareAnti => UNMATCHED, + // Every probe row once, with whether it matched. + JoinType::Existence => return self.tail_exists(shared), }; let rows: Vec = (0..self.state.len()) .filter(|&i| self.state[i] == want) @@ -302,6 +312,35 @@ impl Cursor { let out = Batch::with_weights(record(&shared.schema, columns, m)?, weights); out.map(Some).map_err(|e| internal(e.to_string())) } + + /// Every probe row once plus `exists`: whether some build row matched + /// it. Never NULL: a NULL `=` key matches nothing and reads FALSE. + /// Rows deferred to a spilled partition are skipped here and joined in + /// the grace phase. + fn tail_exists(&self, shared: &Shared) -> Result> { + let mut rows = Vec::new(); + let mut marks: Vec = Vec::new(); + for (i, &state) in self.state.iter().enumerate() { + if state == SKIP { + continue; + } + rows.push(u32::try_from(i).map_err(|_| internal("probe batch too large".to_owned()))?); + marks.push(state == MATCHED); + } + if rows.is_empty() { + return Ok(None); + } + let taken = if rows.len() == self.batch.num_rows() { + self.batch.clone() + } else { + take_batch(&self.batch, &UInt32Array::from(rows))? + }; + let (data, weights) = taken.into_parts(); + let mut columns = data.columns().to_vec(); + columns.push(Arc::new(BooleanArray::from(marks)) as ArrayRef); + let out = Batch::with_weights(record(&shared.schema, columns, data.num_rows())?, weights); + out.map(Some).map_err(|e| internal(e.to_string())) + } } fn overflow() -> KrishivError { @@ -313,6 +352,12 @@ fn overflow() -> KrishivError { .into() } +/// A probe row matched more than once in a single-match left outer join: a +/// correlated scalar subquery returned more than one row. +fn too_many_rows() -> KrishivError { + ScalarSubqueryTooManyRows {}.into() +} + struct ProbeLocal { build: Arc, /// Count probe lengths (`ExecConfig::detailed_metrics`). diff --git a/crates/krishiv-exec/src/physical/instantiate.rs b/crates/krishiv-exec/src/physical/instantiate.rs index d3cca2388..c71bd8259 100644 --- a/crates/krishiv-exec/src/physical/instantiate.rs +++ b/crates/krishiv-exec/src/physical/instantiate.rs @@ -824,6 +824,17 @@ pub(crate) fn hash_join( spill: Option, cell: &mut dyn FnMut(FilterId) -> Result>, ) -> Result { + if j.distinct_build && !j.distinct_build_allowed() { + return Err(internal(format!( + "a distinct build on a {:?} join{}", + j.join_type, + if j.residual.is_some() { + " with a residual condition" + } else { + "" + } + ))); + } let join = HashJoin::try_new(JoinSpec { join_type: j.join_type, probe_schema: Arc::clone(probe), @@ -832,6 +843,7 @@ pub(crate) fn hash_join( build_keys: j.build_keys.clone(), null_safe: j.null_safe.clone(), filter: residual(probe, build, j.residual.as_ref(), env)?, + single_match: j.single_match, spill, })?; if j.runtime_filters.is_empty() { diff --git a/crates/krishiv-exec/src/physical/mod.rs b/crates/krishiv-exec/src/physical/mod.rs index 17c825681..e2ad0bc01 100644 --- a/crates/krishiv-exec/src/physical/mod.rs +++ b/crates/krishiv-exec/src/physical/mod.rs @@ -433,12 +433,36 @@ pub struct HashJoinPlan { pub null_safe: Vec, /// The non-equality conditions. pub residual: Option, + /// A left outer join that fails when a probe row matches more than + /// once (a correlated scalar subquery without an aggregate). Only + /// `LeftOuter` joins set it. + pub single_match: bool, + /// The build's rows are never output and the join reads nothing of + /// them but their keys (a left semi, left anti or existence join + /// without a residual condition), so the build may keep each key once + /// (design P2-T166 §3.2). An executor that ignores it stores every row + /// and returns the same result; this one does so for now. + pub distinct_build: bool, /// The runtime filters the build publishes. pub runtime_filters: Vec, /// Broadcast or partitioned. pub distribution: JoinDistribution, } +impl HashJoinPlan { + /// Whether [`distinct_build`](Self::distinct_build) may be set: the + /// join outputs no build rows and reads no build column but the keys + /// (a left semi, left anti or existence join without a residual). + #[must_use] + pub fn distinct_build_allowed(&self) -> bool { + self.residual.is_none() + && matches!( + self.join_type, + JoinType::LeftSemi | JoinType::LeftAnti | JoinType::Existence + ) + } +} + /// A nested-loop join over every pair of rows. #[derive(Debug, Clone, PartialEq)] pub struct NestedLoopPlan { @@ -504,8 +528,8 @@ plan_data!(struct SubqueryPlan { id, kind, data_type, plan }); plan_data!(struct AggregatePlan { group_by, aggregates, distinct, output }); plan_data!(struct Residual { layout, conjuncts }); plan_data!(struct HashJoinPlan { - probe, build, join_type, probe_keys, build_keys, null_safe, residual, runtime_filters, - distribution + probe, build, join_type, probe_keys, build_keys, null_safe, residual, single_match, + distinct_build, runtime_filters, distribution }); plan_data!(struct NestedLoopPlan { probe, build, join_type, residual }); plan_data!(struct PhysicalPlan { root, schema, explain, warnings, env }); diff --git a/crates/krishiv-exec/src/planner/join.rs b/crates/krishiv-exec/src/planner/join.rs index 8982d6808..4a3695259 100644 --- a/crates/krishiv-exec/src/planner/join.rs +++ b/crates/krishiv-exec/src/planner/join.rs @@ -138,20 +138,25 @@ impl Planner<'_> { (&logical.0, &logical.1) }; let build_rows = self.estimated_rows(build_plan); + // No logical join is a scalar subquery's single-match join yet + // (design P2-T115 unit 2); orientation already refuses to flip one. + let single_match = false; // Which side the statistics say to build (design P2-T166). The // joins that build their preserved side do not run yet: the plan // keeps the right build and shows the preference. let preferred = if choice.distribution.is_none() { let probe_stats = self.input_stats(probe_plan, &probe.schema); let build_stats = self.input_stats(build_plan, &build.schema); - let other = orient(join_type, &probe_stats, &build_stats); + let other = orient(join_type, single_match, &probe_stats, &build_stats); (other != join_type).then_some((other, probe_stats, build_stats)) } else { None }; // A hint decides, else the build's estimated size. The null-aware // anti join reads whether the whole build has a NULL key, so its - // build is always broadcast. + // build is always broadcast. An existence join reads only the build + // rows with the probe row's key (its `exists` is never NULL), so, + // like the other equi-joins, it is broadcast or partitioned by size. let distribution = if join_type == JoinType::NullAwareAnti { JoinDistribution::Broadcast } else { @@ -232,6 +237,8 @@ impl Planner<'_> { build_keys, null_safe, residual: filter, + single_match, + distinct_build: false, runtime_filters, distribution, }; diff --git a/crates/krishiv-exec/src/planner/orient.rs b/crates/krishiv-exec/src/planner/orient.rs index 9ebce474e..e491ed696 100644 --- a/crates/krishiv-exec/src/planner/orient.rs +++ b/crates/krishiv-exec/src/planner/orient.rs @@ -16,6 +16,11 @@ //! //! - A null-aware anti join (`NOT IN`) is never reoriented: it reads //! whether the whole subquery side holds a NULL key. +//! - Existence joins are never reoriented: their output is one row per +//! probe row (DuckDB's mark join has no inverse either). +//! - A single-match left outer join (a scalar subquery) is never +//! reoriented: its at-most-one-match check counts the matches of each +//! probe row, which a build-side outer join does not see. //! - Inner joins keep the join orderer's choice. //! - A build the planner would broadcast (at most //! [`BROADCAST_MAX_BYTES`](super::join::BROADCAST_MAX_BYTES)) stays. @@ -71,14 +76,19 @@ impl InputStats { } /// The join a hash join with a right build becomes when it builds its left -/// input instead, if it can. -pub(crate) fn flipped(t: JoinType) -> Option { +/// input instead, if it can; `single_match` is +/// [`HashJoinPlan::single_match`](crate::physical::HashJoinPlan::single_match). +pub(crate) fn flipped(t: JoinType, single_match: bool) -> Option { + if single_match { + return None; + } match t { JoinType::LeftSemi => Some(JoinType::RightSemi), JoinType::LeftAnti => Some(JoinType::RightAnti), JoinType::LeftOuter => Some(JoinType::RightOuter), JoinType::Inner | JoinType::NullAwareAnti + | JoinType::Existence | JoinType::RightSemi | JoinType::RightAnti | JoinType::RightOuter => None, @@ -88,8 +98,13 @@ pub(crate) fn flipped(t: JoinType) -> Option { /// The join type of a hash join whose right input builds by default, given /// its inputs: `join_type` itself (build right), or the variant that builds /// the left input. -pub(crate) fn orient(join_type: JoinType, left: &InputStats, right: &InputStats) -> JoinType { - let Some(other) = flipped(join_type) else { +pub(crate) fn orient( + join_type: JoinType, + single_match: bool, + left: &InputStats, + right: &InputStats, +) -> JoinType { + let Some(other) = flipped(join_type, single_match) else { return join_type; }; if right.bytes <= BROADCAST_MAX_BYTES { @@ -135,9 +150,9 @@ mod tests { (JoinType::LeftAnti, JoinType::RightAnti), (JoinType::LeftOuter, JoinType::RightOuter), ] { - assert_eq!(orient(t, &small, &large), want, "{t:?}"); + assert_eq!(orient(t, false, &small, &large), want, "{t:?}"); // The other way round keeps the right build. - assert_eq!(orient(t, &large, &small), t, "{t:?}"); + assert_eq!(orient(t, false, &large, &small), t, "{t:?}"); } } @@ -147,24 +162,52 @@ mod tests { for t in [ JoinType::Inner, JoinType::NullAwareAnti, + JoinType::Existence, JoinType::RightSemi, JoinType::RightAnti, JoinType::RightOuter, ] { - assert_eq!(orient(t, &small, &large), t, "{t:?}"); - assert_eq!(flipped(t), None, "{t:?}"); + assert_eq!(orient(t, false, &small, &large), t, "{t:?}"); + assert_eq!(flipped(t, false), None, "{t:?}"); } } + /// A single-match left outer join (a scalar subquery) keeps its right + /// build however lopsided its inputs: a right outer join cannot check + /// that each probe row matches at most once. + #[test] + fn a_single_match_left_outer_join_keeps_its_build() { + let (small, large) = (known(1.0 * MIB), known(5000.0 * MIB)); + assert_eq!( + orient(JoinType::LeftOuter, false, &small, &large), + JoinType::RightOuter + ); + assert_eq!( + orient(JoinType::LeftOuter, true, &small, &large), + JoinType::LeftOuter + ); + assert_eq!(flipped(JoinType::LeftOuter, true), None); + } + #[test] fn a_broadcastable_right_input_stays_the_build() { let left = known(1.0); assert_eq!( - orient(JoinType::LeftSemi, &left, &known(BROADCAST_MAX_BYTES)), + orient( + JoinType::LeftSemi, + false, + &left, + &known(BROADCAST_MAX_BYTES) + ), JoinType::LeftSemi ); assert_eq!( - orient(JoinType::LeftSemi, &left, &known(BROADCAST_MAX_BYTES + 1.0)), + orient( + JoinType::LeftSemi, + false, + &left, + &known(BROADCAST_MAX_BYTES + 1.0) + ), JoinType::RightSemi ); } @@ -174,11 +217,11 @@ mod tests { let right = known(100.0 * MIB); // Exactly the factor is a tie: the right input stays the build. assert_eq!( - orient(JoinType::LeftAnti, &known(50.0 * MIB), &right), + orient(JoinType::LeftAnti, false, &known(50.0 * MIB), &right), JoinType::LeftAnti ); assert_eq!( - orient(JoinType::LeftAnti, &known(49.0 * MIB), &right), + orient(JoinType::LeftAnti, false, &known(49.0 * MIB), &right), JoinType::RightAnti ); // A default estimate on either side needs a ratio above 8. @@ -188,11 +231,11 @@ mod tests { ] { let right = stats(100.0 * MIB, r); assert_eq!( - orient(JoinType::LeftSemi, &stats(20.0 * MIB, l), &right), + orient(JoinType::LeftSemi, false, &stats(20.0 * MIB, l), &right), JoinType::LeftSemi ); assert_eq!( - orient(JoinType::LeftSemi, &stats(12.0 * MIB, l), &right), + orient(JoinType::LeftSemi, false, &stats(12.0 * MIB, l), &right), JoinType::RightSemi ); } @@ -203,12 +246,12 @@ mod tests { for bad in [f64::NAN, f64::INFINITY] { let left = known(bad); assert_eq!( - orient(JoinType::LeftSemi, &left, &known(100.0 * MIB)), + orient(JoinType::LeftSemi, false, &left, &known(100.0 * MIB)), JoinType::LeftSemi ); } assert_eq!( - orient(JoinType::LeftSemi, &known(1.0), &known(f64::NAN)), + orient(JoinType::LeftSemi, false, &known(1.0), &known(f64::NAN)), JoinType::LeftSemi ); } @@ -239,11 +282,14 @@ mod tests { let r = stats(b, sources[(i / 2) % 2]); cases.push((t, l, r)); } - let first: Vec<_> = cases.iter().map(|(t, l, r)| orient(*t, l, r)).collect(); + let first: Vec<_> = cases + .iter() + .map(|(t, l, r)| orient(*t, false, l, r)) + .collect(); let reversed: Vec<_> = cases .iter() .rev() - .map(|(t, l, r)| orient(*t, l, r)) + .map(|(t, l, r)| orient(*t, false, l, r)) .collect(); assert!(first.iter().eq(reversed.iter().rev())); assert!(first.iter().any(|t| matches!( diff --git a/crates/krishiv-exec/tests/hash_join/existence.rs b/crates/krishiv-exec/tests/hash_join/existence.rs new file mode 100644 index 000000000..24b299736 --- /dev/null +++ b/crates/krishiv-exec/tests/hash_join/existence.rs @@ -0,0 +1,915 @@ +// SPDX-License-Identifier: Apache-2.0 +//! Existence joins (Spark's `ExistenceJoin`) and single-match left outer +//! joins (P2-T115). +//! +//! The reference model is Spark's definition, written independently of the +//! executor: a probe row's `exists` is TRUE iff some build row makes the +//! whole join condition (key equalities AND the residual) TRUE under SQL +//! three-valued logic, else FALSE; it is never NULL. The cases replayed +//! from Spark 4.2.0 (results recorded on Spark 4.2.0, ANSI on and off +//! alike, and from `tests/spark-golden`) run each statement through the +//! join shape Spark plans for it (its `EXPLAIN`). + +use std::sync::Arc; + +use arrow_array::cast::AsArray; +use arrow_array::types::Int64Type; +use arrow_array::{Array, BooleanArray, Int64Array, RecordBatch}; +use arrow_schema::{DataType, Field, Schema, SchemaRef}; +use krishiv_common::error::ErrorClass; +use krishiv_exec::{ + ExecConfig, ExecNode, HashJoin, JoinSpec, JoinType, NestedLoopJoin, NestedLoopJoinNode, + NestedLoopSpec, Predicate, Result, +}; +use krishiv_memory::SpillScope; +use krishiv_types::batch::{Batch, Weights}; +use proptest::prelude::*; + +use super::util::*; + +/// One row: a join key `k` and a value `v` the residual reads. +#[derive(Clone, Copy, Debug)] +struct Row { + k: Option, + v: Option, +} + +fn schema() -> SchemaRef { + Arc::new(Schema::new(vec![ + Field::new("k", DataType::Int64, true), + Field::new("v", DataType::Int64, true), + ])) +} + +fn data(rows: &[Row]) -> RecordBatch { + RecordBatch::try_new( + schema(), + vec![ + Arc::new(Int64Array::from( + rows.iter().map(|r| r.k).collect::>(), + )), + Arc::new(Int64Array::from( + rows.iter().map(|r| r.v).collect::>(), + )), + ], + ) + .unwrap() +} + +fn r(k: Option, v: Option) -> Row { + Row { k, v } +} + +/// `probe.v < build.v`, NULL when either side is NULL or their sum is a +/// multiple of 7. +#[derive(Debug)] +struct VLess; + +impl Predicate for VLess { + fn evaluate(&self, batch: &RecordBatch) -> Result { + let p = batch.column(1).as_primitive::(); + let b = batch.column(3).as_primitive::(); + Ok(p.iter() + .zip(b.iter()) + .map(|(p, b)| match (p, b) { + (Some(p), Some(b)) => ((p + b) % 7 != 0).then_some(p < b), + _ => None, + }) + .collect()) + } +} + +/// Spark's null-aware `NOT IN` condition `(p = b) OR isnull(p = b)` over +/// pair columns `p` and `b` (probe columns, then build columns): TRUE when +/// either side is NULL or they are equal, else FALSE. +#[derive(Debug)] +struct NullAwareEq { + p: usize, + b: usize, +} + +impl Predicate for NullAwareEq { + fn evaluate(&self, batch: &RecordBatch) -> Result { + let p = batch.column(self.p).as_primitive::(); + let b = batch.column(self.b).as_primitive::(); + Ok(p.iter() + .zip(b.iter()) + .map(|(p, b)| Some(p.is_none() || b.is_none() || p == b)) + .collect()) + } +} + +/// SQL `x AND y` over three-valued booleans. +fn and3(x: Option, y: Option) -> Option { + match (x, y) { + (Some(false), _) | (_, Some(false)) => Some(false), + (Some(true), Some(true)) => Some(true), + _ => None, + } +} + +#[derive(Clone, Debug)] +struct Case { + join_type: JoinType, + null_safe: bool, + filter: bool, + single_match: bool, + row_layout: bool, +} + +impl Case { + fn existence(null_safe: bool, filter: bool, row_layout: bool) -> Self { + Self { + join_type: JoinType::Existence, + null_safe, + filter, + single_match: false, + row_layout, + } + } + + fn single(filter: bool, row_layout: bool) -> Self { + Self { + join_type: JoinType::LeftOuter, + null_safe: false, + filter, + single_match: true, + row_layout, + } + } + + fn spec(&self, spill: Option) -> JoinSpec { + JoinSpec { + join_type: self.join_type, + probe_schema: schema(), + build_schema: schema(), + probe_keys: vec![0], + build_keys: vec![0], + null_safe: vec![self.null_safe], + filter: self.filter.then(|| Arc::new(VLess) as Arc), + single_match: self.single_match, + spill, + } + } + + /// The join condition of a pair, in SQL three-valued logic: `p.k = b.k` + /// (or `p.k <=> b.k`) AND (with the residual) `VLess`. + fn condition(&self, p: &Row, b: &Row) -> Option { + let key = match (p.k, b.k) { + (Some(x), Some(y)) => Some(x == y), + (x, y) if self.null_safe => Some(x == y), + _ => None, + }; + let residual = if self.filter { + match (p.v, b.v) { + (Some(x), Some(y)) => ((x + y) % 7 != 0).then_some(x < y), + _ => None, + } + } else { + Some(true) + }; + and3(key, residual) + } + + /// Spark's `exists` of `p`: some build row makes the condition TRUE. + fn exists(&self, p: &Row, build: &[Row]) -> bool { + build.iter().any(|b| self.condition(p, b) == Some(true)) + } + + /// The reference existence join, rendered like [`render`]. + fn reference_existence(&self, probe: &[Row], build: &[Row]) -> Vec { + let cell = |x: &Option| x.map_or("∅".to_owned(), |v| v.to_string()); + let mut out: Vec = probe + .iter() + .map(|p| format!("{}|{}|{}", cell(&p.k), cell(&p.v), self.exists(p, build))) + .collect(); + out.sort(); + out + } + + /// The reference single-match join (at most one match per probe row, + /// as the tests arrange), rendered like [`render`]. + fn reference_single(&self, probe: &[Row], build: &[Row]) -> Vec { + let cell = |x: &Option| x.map_or("∅".to_owned(), |v| v.to_string()); + let mut out: Vec = probe + .iter() + .map(|p| { + let hits: Vec<&Row> = build + .iter() + .filter(|b| self.condition(p, b) == Some(true)) + .collect(); + assert!(hits.len() <= 1, "the reference holds one match at most"); + match hits.as_slice() { + [b] => format!( + "{}|{}|{}|{}", + cell(&p.k), + cell(&p.v), + cell(&b.k), + cell(&b.v) + ), + _ => format!("{}|{}|∅|∅", cell(&p.k), cell(&p.v)), + } + }) + .collect(); + out.sort(); + out + } + + fn reference(&self, probe: &[Row], build: &[Row]) -> Vec { + if self.single_match { + self.reference_single(probe, build) + } else { + self.reference_existence(probe, build) + } + } +} + +/// Runs `case` over `probe` ⋈ `build`. +async fn run_case( + case: &Case, + probe: &[Row], + build: &[Row], + batch_rows: usize, + threads: usize, + limit: usize, +) -> (Result>, Arc) { + let root = root(limit); + let (_manager, scope) = spill(); + let config = ExecConfig { + batch_size: if probe.len() < 1000 { 5 } else { 1024 }, + join_row_layout: case.row_layout, + ..ExecConfig::default() + }; + let q = query(&root, 1, config); + let plan = join( + case.spec(Some(scope.clone())), + scan(&schema(), batches(&data(build), batch_rows)), + scan(&schema(), batches(&data(probe), batch_rows)), + ); + let out = run(&runner_for(threads, limit), plan, &q) + .await + .map(|b| render(&b)); + assert_released(&root); + assert_eq!(spill_files(&scope), 0, "spill files are deleted"); + (out, q) +} + +/// Every combination of match/no match, NULL or non-NULL probe key, NULL +/// key in the build or not, and an empty build, with and without the +/// residual condition, `=` and `<=>` keys, in both key formats. +#[tokio::test] +async fn existence_matches_the_reference_on_edge_cases() { + let probe = vec![ + r(Some(1), Some(1)), + r(Some(1), Some(90)), + r(Some(2), Some(3)), + r(Some(3), None), + r(None, Some(4)), + r(None, None), + ]; + // A NULL value exercises a NULL residual on an otherwise matching pair + // (`Some(3)` meets `None`); the sums that are multiples of 7 do too. + let builds: &[&[Row]] = &[ + &[r(Some(1), Some(8)), r(Some(2), Some(4))], + &[r(Some(1), Some(8)), r(None, Some(5))], + &[r(Some(1), None)], + &[r(Some(9), Some(9))], + &[r(None, Some(5))], + &[], + ]; + for null_safe in [false, true] { + for filter in [false, true] { + for row_layout in [false, true] { + let case = Case::existence(null_safe, filter, row_layout); + for build in builds { + let (got, _) = run_case(&case, &probe, build, 2, 2, 64 * MIB).await; + assert_eq!( + got.unwrap(), + case.reference(&probe, build), + "{case:?} build {build:?}" + ); + } + } + } + } +} + +/// `exists` is never NULL: a NULL `=` key on either side, a NULL key in +/// the build and a NULL residual all read FALSE, never NULL (Spark's +/// `ExistenceJoin`; the T115 design note's three-valued mark is not what +/// Spark returns). +#[tokio::test] +async fn exists_is_never_null() { + let probe = [r(Some(2), Some(1)), r(None, Some(1)), r(Some(3), None)]; + let build = [r(Some(1), Some(1)), r(None, Some(1)), r(Some(3), Some(5))]; + for filter in [false, true] { + let case = Case::existence(false, filter, false); + let (got, _) = run_case(&case, &probe, &build, 1, 1, 64 * MIB).await; + let got = got.unwrap(); + assert!(got.iter().all(|row| !row.ends_with('∅')), "{got:?}"); + let expect = if filter { + ["2|1|false", "3|∅|false", "∅|1|false"] + } else { + ["2|1|false", "3|∅|true", "∅|1|false"] + }; + assert_eq!(got, expect, "{case:?}"); + } +} + +/// The specs the executor refuses: single match on a join that is not +/// left outer. Existence joins run by hash over any keys (null-safe or +/// not, with a residual) and by nested loops. +#[test] +fn existence_and_single_match_specs() { + let schema = schema(); + let spec = |join_type, single_match, keys: usize| JoinSpec { + join_type, + probe_schema: Arc::clone(&schema), + build_schema: Arc::clone(&schema), + probe_keys: vec![0; keys], + build_keys: vec![0; keys], + null_safe: vec![true; keys], + filter: Some(Arc::new(VLess) as Arc), + single_match, + spill: None, + }; + assert!(HashJoin::try_new(spec(JoinType::Existence, false, 2)).is_ok()); + let nested = NestedLoopJoin::try_new(NestedLoopSpec { + join_type: JoinType::Existence, + probe_schema: Arc::clone(&schema), + build_schema: Arc::clone(&schema), + filter: None, + }) + .unwrap(); + assert_eq!( + nested.schema().field(2), + &Field::new("exists", DataType::Boolean, false) + ); + for join_type in [ + JoinType::Inner, + JoinType::LeftSemi, + JoinType::LeftAnti, + JoinType::NullAwareAnti, + JoinType::Existence, + ] { + let mut s = spec(join_type, true, 1); + s.filter = None; + s.null_safe = vec![false]; + let e = HashJoin::try_new(s).err().unwrap(); + assert_eq!(e.error_class(), ErrorClass::InternalError, "{join_type:?}"); + } +} + +/// Zero or one match: a single-match join emits what a left outer join +/// does. +#[tokio::test] +async fn single_match_with_zero_or_one_match_is_a_left_outer_join() { + for filter in [false, true] { + for row_layout in [false, true] { + let case = Case::single(filter, row_layout); + let probe = [r(Some(1), Some(1)), r(Some(2), Some(3)), r(None, Some(4))]; + let build = [r(Some(1), Some(8))]; + let (got, _) = run_case(&case, &probe, &build, 2, 2, 64 * MIB).await; + assert_eq!(got.unwrap(), case.reference(&probe, &build), "{case:?}"); + } + } +} + +/// Two matches fail the query with `SCALAR_SUBQUERY_TOO_MANY_ROWS` +/// (SQLSTATE 21000), with and without a residual condition; one match +/// filtered down from two is fine. +#[tokio::test] +async fn single_match_with_two_matches_fails() { + for filter in [false, true] { + let case = Case::single(filter, false); + // `Some(1)` meets both build rows; with the filter both pairs hold + // (`1 < 8`, `1 < 9`, and neither sum is a multiple of 7). + let probe = [r(Some(1), Some(1))]; + let build = [r(Some(1), Some(8)), r(Some(1), Some(9))]; + let root = root(64 * MIB); + let q = query(&root, 1, ExecConfig::default()); + let plan = join( + case.spec(None), + scan(&schema(), vec![Batch::new(data(&build))]), + scan(&schema(), vec![Batch::new(data(&probe))]), + ); + let e = run(&runner(2), plan, &q).await.unwrap_err(); + assert_eq!( + e.error_class(), + ErrorClass::ScalarSubqueryTooManyRows, + "{e}" + ); + assert_eq!(e.sql_state(), Some("21000")); + assert!( + e.to_string() + .contains("More than one row returned by a subquery used as an expression"), + "{e}" + ); + assert_released(&root); + } + let case = Case::single(true, false); + // `1 < 0` fails, `1 < 8` holds. + let probe = [r(Some(1), Some(1))]; + let build = [r(Some(1), Some(0)), r(Some(1), Some(8))]; + let (got, _) = run_case(&case, &probe, &build, 1, 1, 64 * MIB).await; + assert_eq!(got.unwrap(), case.reference(&probe, &build)); + // A NULL residual on the second pair is not a match either. + let build = [r(Some(1), Some(8)), r(Some(1), None)]; + let (got, _) = run_case(&case, &probe, &build, 1, 1, 64 * MIB).await; + assert_eq!(got.unwrap(), case.reference(&probe, &build)); +} + +/// Two matches with spill forced fail the same way, releasing memory and +/// deleting spill files. +#[tokio::test(start_paused = true)] +async fn single_match_with_two_matches_fails_when_spilling() { + let build: Vec = (0..20_000) + .map(|i| r(Some(i % 9_000), Some(i % 50))) + .chain([r(Some(7), Some(1)), r(Some(7), Some(2))]) + .collect(); + let probe = [r(Some(7), Some(0))]; + let case = Case::single(false, false); + let root = root(1 << 20); + let (_manager, scope) = spill(); + let q = query(&root, 1, ExecConfig::default()); + let plan = join( + case.spec(Some(scope.clone())), + scan(&schema(), batches(&data(&build), 512)), + scan(&schema(), batches(&data(&probe), 512)), + ); + let e = run(&runner_for(2, 1 << 20), plan, &q).await.unwrap_err(); + assert_eq!(e.error_class(), ErrorClass::ScalarSubqueryTooManyRows); + assert_released(&root); + assert_eq!(spill_files(&scope), 0); +} + +fn skewed(rows: usize, keys: i64, null_every: i64) -> Vec { + (0..rows) + .map(|i| { + let i = i64::try_from(i).unwrap(); + let k = (i % null_every != 0).then_some(i % keys); + r(k, Some(i % 50)) + }) + .collect() +} + +/// A tiny budget forces the existence join through grace spill +/// (partitions, deferred probe rows, repartitioning): the answer is the +/// reference's, in both key formats. +#[tokio::test(start_paused = true)] +async fn spilling_existence_joins_match_the_reference() { + let build = skewed(30_000, 20_000, 17); + let mut probe = skewed(40_000, 25_000, 13); + probe.extend((0..50).map(|i| r(None, Some(i)))); + for filter in [false, true] { + for row_layout in [false, true] { + let case = Case::existence(false, filter, row_layout); + let expect = case.reference(&probe, &build); + let (got, q) = run_case(&case, &probe, &build, 1024, 3, 768 << 10).await; + assert!(got.unwrap() == expect, "{case:?}: grace result differs"); + assert!( + metric(&q, "HashJoinBuild", "spilled_partitions") > 0, + "{case:?}" + ); + assert!(metric(&q, "HashJoinProbe", "grace_partitions") > 0); + // NULL `=` keys never match: those build rows are not stored. + let stored = build.iter().filter(|r| r.k.is_some()).count() as u64; + assert_eq!(metric(&q, "HashJoinBuild", "build_rows"), stored); + assert_eq!(metric(&q, "HashJoinProbe", "probe_rows"), 40_050); + } + } +} + +/// A weighted build side is rejected for existence and single-match joins, +/// as for the other non-inner join types; probe weights pass through. +#[tokio::test] +async fn weights_pass_through_or_are_rejected_like_the_other_joins() { + let probe = [r(Some(1), Some(1)), r(Some(2), Some(2))]; + let build = [r(Some(1), Some(3))]; + let cases = [ + Case::existence(false, false, false), + Case::single(false, false), + ]; + for case in &cases { + let root = root(64 * MIB); + let q = query(&root, 1, ExecConfig::default()); + let plan = join( + case.spec(None), + scan( + &schema(), + vec![Batch::with_weights(data(&build), Weights::Const(2)).unwrap()], + ), + scan(&schema(), vec![Batch::new(data(&probe))]), + ); + let e = run(&runner(1), plan, &q).await.unwrap_err(); + assert_eq!(e.error_class(), ErrorClass::KrishivUnsupportedFeature); + assert_released(&root); + } + for (case, expect) in cases.iter().zip([ + vec!["1|1|true|*2", "2|2|false|*5"], + vec!["1|1|1|3|*2", "2|2|∅|∅|*5"], + ]) { + let root = root(64 * MIB); + let q = query(&root, 1, ExecConfig::default()); + let plan = join( + case.spec(None), + scan(&schema(), vec![Batch::new(data(&build))]), + scan( + &schema(), + vec![ + Batch::with_weights( + data(&probe), + Weights::Column(Int64Array::from(vec![2, 5])), + ) + .unwrap(), + ], + ), + ); + let out = run(&runner(1), plan, &q).await.unwrap(); + assert_eq!(render(&out), expect, "{case:?}"); + assert_released(&root); + } +} + +fn row(keys: i64, nulls: u32) -> impl Strategy { + let k = prop_oneof![1 => Just(None), nulls => (0i64..keys).prop_map(Some)]; + (k, prop::option::of(0i64..20)).prop_map(|(k, v)| Row { k, v }) +} + +/// The partition of a key under a co-partitioning by `k` into `n` parts. +/// NULL keys of the probe and of the build go to different partitions: an +/// existence join needs no build-wide fact, so where they go is irrelevant. +fn part(k: Option, n: i64, probe: bool) -> i64 { + match k { + Some(k) => k.rem_euclid(n), + None if probe => 0, + None => n - 1, + } +} + +proptest! { + #![proptest_config(ProptestConfig { cases: 48, ..ProptestConfig::default() })] + /// Random data with NULLs, with and without the residual condition, + /// `=` and `<=>` keys, in both key formats: always the reference. + #[test] + fn random_existence_matches_the_reference( + null_safe in any::(), + filter in any::(), + row_layout in any::(), + probe in prop::collection::vec(row(6, 4), 0..40), + build in prop::collection::vec(row(6, 4), 0..40), + batch_rows in 1usize..17, + threads in 1usize..4, + ) { + let case = Case::existence(null_safe, filter, row_layout); + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + let (got, _) = rt.block_on(run_case(&case, &probe, &build, batch_rows, threads, 64 * MIB)); + prop_assert_eq!(got.unwrap(), case.reference(&probe, &build)); + } + + /// A partitioned existence join (probe and build co-partitioned by the + /// key, each partition joined on its own, as tasks of a partitioned + /// join are) returns what one join over all rows returns. + #[test] + fn partitioned_existence_equals_one_join( + filter in any::(), + parts in 1i64..5, + probe in prop::collection::vec(row(8, 3), 0..40), + build in prop::collection::vec(row(8, 3), 0..40), + ) { + let case = Case::existence(false, filter, false); + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + let mut got = Vec::new(); + for p in 0..parts { + let probe: Vec = + probe.iter().filter(|r| part(r.k, parts, true) == p).copied().collect(); + let build: Vec = + build.iter().filter(|r| part(r.k, parts, false) == p).copied().collect(); + let (out, _) = rt.block_on(run_case(&case, &probe, &build, 4, 2, 64 * MIB)); + got.extend(out.unwrap()); + } + got.sort(); + prop_assert_eq!(got, case.reference(&probe, &build)); + } +} + +// --------------------------------------------------------------------------- +// Spark 4.2.0 replays. +// --------------------------------------------------------------------------- + +/// Rows of nullable Int64 columns named `names`. +fn table(names: &[&str], rows: &[&[Option]]) -> (SchemaRef, RecordBatch) { + let schema = Arc::new(Schema::new( + names + .iter() + .map(|n| Field::new(*n, DataType::Int64, true)) + .collect::>(), + )); + let columns = (0..names.len()) + .map(|c| { + Arc::new(Int64Array::from( + rows.iter().map(|r| r[c]).collect::>(), + )) as _ + }) + .collect(); + let batch = RecordBatch::try_new_with_options( + Arc::clone(&schema), + columns, + &arrow_array::RecordBatchOptions::new().with_row_count(Some(rows.len())), + ) + .unwrap(); + (schema, batch) +} + +/// How Spark plans the statement (its `EXPLAIN`). +enum Shape { + /// `BroadcastHashJoin [probe keys], [build keys], ExistenceJoin`, with + /// an optional `(p = b) OR isnull(p = b)` residual over pair columns. + Hash { + probe_keys: Vec, + build_keys: Vec, + null_aware: Option<(usize, usize)>, + }, + /// `BroadcastNestedLoopJoin ExistenceJoin, (p = b) OR isnull(p = b)`. + NestedNullAware(usize, usize), +} + +/// The `exists` column of `probe` joined with `build` in `shape`, in probe +/// row order (sorted by the first probe column, an id). +async fn exists_of( + shape: &Shape, + probe: &(SchemaRef, RecordBatch), + build: &(SchemaRef, RecordBatch), +) -> Vec<(i64, bool)> { + let root = root(64 * MIB); + let q = query(&root, 1, ExecConfig::default()); + let probe_scan = scan(&probe.0, vec![Batch::new(probe.1.clone())]); + let build_scan = scan(&build.0, vec![Batch::new(build.1.clone())]); + let plan: Arc = match shape { + Shape::Hash { + probe_keys, + build_keys, + null_aware, + } => join( + JoinSpec { + join_type: JoinType::Existence, + probe_schema: Arc::clone(&probe.0), + build_schema: Arc::clone(&build.0), + probe_keys: probe_keys.clone(), + build_keys: build_keys.clone(), + null_safe: vec![false; probe_keys.len()], + filter: null_aware.map(|(p, b)| Arc::new(NullAwareEq { p, b }) as _), + single_match: false, + spill: None, + }, + build_scan, + probe_scan, + ), + Shape::NestedNullAware(p, b) => Arc::new(NestedLoopJoinNode { + join: NestedLoopJoin::try_new(NestedLoopSpec { + join_type: JoinType::Existence, + probe_schema: Arc::clone(&probe.0), + build_schema: Arc::clone(&build.0), + filter: Some(Arc::new(NullAwareEq { p: *p, b: *b })), + }) + .unwrap(), + build: build_scan, + probe: probe_scan, + }), + }; + let out = run(&runner(2), plan, &q).await.unwrap(); + assert_released(&root); + let mut rows = Vec::new(); + for b in &out { + let b = b.data(); + let id = b.column(0).as_primitive::(); + let exists = b.column(b.num_columns() - 1).as_boolean(); + assert_eq!(exists.null_count(), 0, "exists is never NULL"); + for i in 0..b.num_rows() { + rows.push((id.value(i), exists.value(i))); + } + } + rows.sort_unstable(); + rows +} + +/// The ids and their expected values, `NOT` applied where the statement +/// negates the existence join (`NOT IN`). +fn expect(values: &[bool]) -> Vec<(i64, bool)> { + (1..).zip(values.iter().copied()).collect() +} + +fn negated(rows: Vec<(i64, bool)>) -> Vec<(i64, bool)> { + rows.into_iter().map(|(id, e)| (id, !e)).collect() +} + +/// Results recorded on Spark 4.2.0 (2026-10-25, `spark.sql.ansi.enabled` +/// true and false alike) over +/// `t(id, k) = (1,1), (2,2), (3,NULL), (4,4)`, `un(k) = (1), (NULL)`, +/// `nn(k) = (1), (2)`, `e(k)` empty and +/// `c(g, k) = (1,1), (2,NULL), (4,9), (3,5)`, each run in the shape of +/// Spark's plan. The same values come out in the SELECT list, under `OR` +/// in WHERE, in `CASE` and in a join condition: Spark plans every one as an +/// `ExistenceJoin`, whose `exists` is never NULL. +#[tokio::test] +async fn spark_in_and_not_in_outside_a_filter_conjunct() { + let none = None; + let t = table( + &["id", "k"], + &[ + &[Some(1), Some(1)], + &[Some(2), Some(2)], + &[Some(3), none], + &[Some(4), Some(4)], + ], + ); + let un = table(&["k"], &[&[Some(1)], &[none]]); + let nn = table(&["k"], &[&[Some(1)], &[Some(2)]]); + let e = table(&["k"], &[]); + let c = table( + &["g", "k"], + &[ + &[Some(1), Some(1)], + &[Some(2), none], + &[Some(4), Some(9)], + &[Some(3), Some(5)], + ], + ); + // `k IN (SELECT k FROM s)`: BroadcastHashJoin [k], [k], ExistenceJoin. + let in_shape = Shape::Hash { + probe_keys: vec![1], + build_keys: vec![0], + null_aware: None, + }; + // `k NOT IN (SELECT k FROM s)`: NOT of BroadcastNestedLoopJoin + // ExistenceJoin, (k = s.k) OR isnull(k = s.k). + let not_in_shape = Shape::NestedNullAware(1, 2); + for (s, want_in, want_not_in) in [ + // SELECT id, k IN (SELECT k FROM un): 1 true, 2 false, 3 false, + // 4 false; NOT IN: all false. + (&un, [true, false, false, false], [false; 4]), + // nn: IN 1 true, 2 true, 3 false, 4 false; NOT IN only 4 true. + (&nn, [true, true, false, false], [false, false, false, true]), + // e: IN all false; NOT IN all true (NULL probe key too). + (&e, [false; 4], [true; 4]), + ] { + assert_eq!(exists_of(&in_shape, &t, s).await, expect(&want_in)); + assert_eq!( + negated(exists_of(¬_in_shape, &t, s).await), + expect(&want_not_in) + ); + } + // `k IN (SELECT c.k FROM c WHERE c.g = t.id)`: + // BroadcastHashJoin [k, id], [k, g], ExistenceJoin. 1 true, else false. + let corr_in = Shape::Hash { + probe_keys: vec![1, 0], + build_keys: vec![1, 0], + null_aware: None, + }; + assert_eq!( + exists_of(&corr_in, &t, &c).await, + expect(&[true, false, false, false]) + ); + // `k NOT IN (SELECT c.k FROM c WHERE c.g = t.id)`: NOT of + // BroadcastHashJoin [id], [g], ExistenceJoin, + // (k = c.k) OR isnull(k = c.k). Only 4 true: 2 meets a NULL `c.k`, 3 + // has a NULL `k`. + let corr_not_in = Shape::Hash { + probe_keys: vec![0], + build_keys: vec![0], + null_aware: Some((1, 3)), + }; + assert_eq!( + negated(exists_of(&corr_not_in, &t, &c).await), + expect(&[false, false, false, true]) + ); + // `EXISTS (SELECT 1 FROM c WHERE c.g = t.id AND c.k = t.k)`: 1 true. + assert_eq!( + exists_of(&corr_in, &t, &c).await, + expect(&[true, false, false, false]) + ); +} + +/// `tests/spark-golden/v4.2.0/results/subquery/in-subquery/in-null-semantics.sql.out` +/// (`spark.sql.legacy.nullInEmptyListBehavior = false`): over +/// `t(c) = (1), (NULL)` and an empty `t_empty(e)`, +/// `c in (select e from t_empty)` is false, false and +/// `c not in (select e from t_empty)` is true, true. +#[tokio::test] +async fn spark_golden_in_null_semantics() { + let t = table(&["c"], &[&[Some(1)], &[None]]); + let t_empty = table(&["e"], &[]); + // Both rows expect the same value, so their order does not matter. + let in_shape = Shape::Hash { + probe_keys: vec![0], + build_keys: vec![0], + null_aware: None, + }; + let rows = exists_of(&in_shape, &t, &t_empty).await; + assert_eq!(rows.iter().map(|r| r.1).collect::>(), [false, false]); + let rows = negated(exists_of(&Shape::NestedNullAware(0, 1), &t, &t_empty).await); + assert_eq!(rows.iter().map(|r| r.1).collect::>(), [true, true]); +} + +/// `tests/spark-golden/v4.2.0/results/subquery/exists-subquery/exists-outside-filter.sql.out`: +/// `SELECT emp_name, EXISTS (SELECT 1 FROM dept WHERE emp.dept_id = +/// dept.dept_id) FROM emp`, emp 1 to emp 8 by id. +#[tokio::test] +async fn spark_golden_exists_outside_filter() { + let emp = table( + &["id", "dept_id"], + &[ + &[Some(100), Some(10)], + &[Some(100), Some(10)], + &[Some(200), Some(10)], + &[Some(300), Some(20)], + &[Some(400), Some(30)], + &[Some(500), None], + &[Some(600), Some(100)], + &[Some(700), Some(100)], + &[Some(800), Some(70)], + ], + ); + let dept = table( + &["dept_id"], + &[ + &[Some(10)], + &[Some(20)], + &[Some(30)], + &[Some(40)], + &[Some(50)], + &[Some(70)], + ], + ); + let shape = Shape::Hash { + probe_keys: vec![1], + build_keys: vec![0], + null_aware: None, + }; + let rows = exists_of(&shape, &emp, &dept).await; + // emp 1, emp 1, emp 2, emp 3, emp 4: true; emp 5 (NULL dept), emp 6, + // emp 7: false; emp 8: true. + assert_eq!( + rows, + [ + (100, true), + (100, true), + (200, true), + (300, true), + (400, true), + (500, false), + (600, false), + (700, false), + (800, true), + ] + ); +} + +/// The nested-loop existence join over random data equals the reference +/// with the key equality moved into the condition. +#[test] +fn nested_loop_existence_matches_the_reference() { + #[derive(Debug)] + struct KeyAndV; + impl Predicate for KeyAndV { + fn evaluate(&self, batch: &RecordBatch) -> Result { + let pk = batch.column(0).as_primitive::(); + let bk = batch.column(2).as_primitive::(); + let v = VLess.evaluate(batch)?; + Ok((0..batch.num_rows()) + .map(|i| { + let key = + (pk.is_valid(i) && bk.is_valid(i)).then(|| pk.value(i) == bk.value(i)); + let v = v.is_valid(i).then(|| v.value(i)); + and3(key, v) + }) + .collect()) + } + } + let case = Case::existence(false, true, false); + let probe = skewed(300, 40, 7); + let build = skewed(200, 50, 5); + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let got = rt.block_on(async { + let root = root(64 * MIB); + let q = query(&root, 1, ExecConfig::default()); + let plan = Arc::new(NestedLoopJoinNode { + join: NestedLoopJoin::try_new(NestedLoopSpec { + join_type: JoinType::Existence, + probe_schema: schema(), + build_schema: schema(), + filter: Some(Arc::new(KeyAndV)), + }) + .unwrap(), + build: scan(&schema(), batches(&data(&build), 64)), + probe: scan(&schema(), batches(&data(&probe), 64)), + }); + let out = run(&runner(2), plan, &q).await.unwrap(); + assert_released(&root); + render(&out) + }); + assert_eq!(got, case.reference(&probe, &build)); +} diff --git a/crates/krishiv-exec/tests/hash_join/main.rs b/crates/krishiv-exec/tests/hash_join/main.rs index ade6d79d7..186ca02b7 100644 --- a/crates/krishiv-exec/tests/hash_join/main.rs +++ b/crates/krishiv-exec/tests/hash_join/main.rs @@ -7,6 +7,7 @@ #![cfg(not(loom))] #![expect(clippy::unwrap_used, reason = "test helpers sit outside #[test] fns")] +mod existence; mod payload_rows; mod runtime_filter; mod tpch; @@ -103,6 +104,7 @@ impl Case { build_keys: (0..self.keys).collect(), null_safe: self.null_safe.clone(), filter: self.filter.then(|| Arc::new(VLess) as Arc), + single_match: false, spill, } } diff --git a/crates/krishiv-exec/tests/hash_join/payload_rows.rs b/crates/krishiv-exec/tests/hash_join/payload_rows.rs index 64ed96714..8ef5ef190 100644 --- a/crates/krishiv-exec/tests/hash_join/payload_rows.rs +++ b/crates/krishiv-exec/tests/hash_join/payload_rows.rs @@ -125,6 +125,7 @@ async fn run_join( build_keys: vec![0], null_safe: vec![false], filter: None, + single_match: false, spill: Some(scope.clone()), }; let plan = join( diff --git a/crates/krishiv-exec/tests/hash_join/tpch.rs b/crates/krishiv-exec/tests/hash_join/tpch.rs index 1a6ed382b..1e028d775 100644 --- a/crates/krishiv-exec/tests/hash_join/tpch.rs +++ b/crates/krishiv-exec/tests/hash_join/tpch.rs @@ -388,6 +388,7 @@ fn hj( build_keys: keys(&build.schema(), build_keys), null_safe: vec![false; probe_keys.len()], filter: filter.map(|f| Arc::new(Cond(f)) as Arc), + single_match: false, spill: Some(spill.clone()), }; let join = HashJoin::try_new(spec).unwrap(); diff --git a/crates/krishiv-exec/tests/planner/physical.rs b/crates/krishiv-exec/tests/planner/physical.rs index 0b9bba51c..2413b776f 100644 --- a/crates/krishiv-exec/tests/planner/physical.rs +++ b/crates/krishiv-exec/tests/planner/physical.rs @@ -222,6 +222,8 @@ fn inconsistent_plans_are_internal_errors() { build_keys: vec![0], null_safe: vec![false], residual: None, + single_match: false, + distinct_build: false, runtime_filters: vec![JoinFilterPlan { id: FilterId(id), key: 0, diff --git a/crates/krishiv-plan-codec/src/decode.rs b/crates/krishiv-plan-codec/src/decode.rs index 29ea3e51e..3eb69c92d 100644 --- a/crates/krishiv-plan-codec/src/decode.rs +++ b/crates/krishiv-plan-codec/src/decode.rs @@ -556,21 +556,33 @@ impl Decoder { if probe_keys.len() != build_keys.len() || probe_keys.len() != j.null_safe.len() { return Err(corrupt("a hash join whose key lists differ in length")); } - Ok(HashJoinPlan { + let join_type = join_type(j.join_type)?; + if j.single_match && join_type != JoinType::LeftOuter { + return Err(corrupt("single match on a join that is not left outer")); + } + let plan = HashJoinPlan { probe: self.boxed(j.probe)?, build: self.boxed(j.build)?, - join_type: join_type(j.join_type)?, + join_type, probe_keys, build_keys, null_safe: j.null_safe.clone(), residual: self.residual(j.residual.as_ref())?, + single_match: j.single_match, + distinct_build: j.distinct_build, runtime_filters: j .runtime_filters .iter() .map(join_filter) .collect::>()?, distribution: distribution(j.distribution)?, - }) + }; + if plan.distinct_build && !plan.distinct_build_allowed() { + return Err(corrupt( + "a distinct build on a join that outputs or reads build rows", + )); + } + Ok(plan) } fn exchange(&mut self, e: &pb::ExchangeNode) -> Result { @@ -860,6 +872,7 @@ fn join_type(t: i32) -> Result { LeftSemi => JoinType::LeftSemi, LeftAnti => JoinType::LeftAnti, NullAwareAnti => JoinType::NullAwareAnti, + Existence => JoinType::Existence, RightSemi => JoinType::RightSemi, RightAnti => JoinType::RightAnti, RightOuter => JoinType::RightOuter, diff --git a/crates/krishiv-plan-codec/src/encode.rs b/crates/krishiv-plan-codec/src/encode.rs index 19c0579f1..9505f8e55 100644 --- a/crates/krishiv-plan-codec/src/encode.rs +++ b/crates/krishiv-plan-codec/src/encode.rs @@ -352,6 +352,8 @@ impl Encoder { residual: self.residual(join.residual.as_ref())?, runtime_filters: join.runtime_filters.iter().map(join_filter).collect(), distribution: distribution(join.distribution) as i32, + single_match: join.single_match, + distinct_build: join.distinct_build, }) } @@ -536,6 +538,7 @@ fn join_type(t: JoinType) -> pb::JoinType { JoinType::LeftSemi => pb::JoinType::LeftSemi, JoinType::LeftAnti => pb::JoinType::LeftAnti, JoinType::NullAwareAnti => pb::JoinType::NullAwareAnti, + JoinType::Existence => pb::JoinType::Existence, JoinType::RightSemi => pb::JoinType::RightSemi, JoinType::RightAnti => pb::JoinType::RightAnti, JoinType::RightOuter => pb::JoinType::RightOuter, diff --git a/crates/krishiv-plan-codec/src/features.rs b/crates/krishiv-plan-codec/src/features.rs index 9869c9f9b..cc78abd8f 100644 --- a/crates/krishiv-plan-codec/src/features.rs +++ b/crates/krishiv-plan-codec/src/features.rs @@ -114,8 +114,10 @@ macro_rules! registry { // The registry: one line per feature, kept alphabetically sorted by id. // (id, variant, legacy level, kind, meaning) registry! { + (existence_join, ExistenceJoin, POST_MIGRATION_READER_VERSION, Semantic, "The `Existence` join type (Spark's `ExistenceJoin`: `EXISTS` and `IN` outside a filter conjunct), hash or nested loop."), (expand, Expand, 5, Semantic, "The `Expand` node and its `NullOf` project item (ROLLUP, CUBE, GROUPING SETS)."), (join_build_preserved, JoinBuildPreserved, POST_MIGRATION_READER_VERSION, Optimization, "Joins that build their preserved side (`RightSemi`, `RightAnti`, `RightOuter`); without it the planner builds the other side (`LeftSemi`, `LeftAnti`, `LeftOuter`)."), + (join_single_match, JoinSingleMatch, POST_MIGRATION_READER_VERSION, Semantic, "The single-match flag of a `LeftOuter` hash join (a scalar subquery: more than one match per probe row is an error)."), (limit_offset, LimitOffset, 3, Semantic, "A `Limit` that skips leading rows (a nonzero offset)."), (window, Window, 4, Semantic, "The `Window` node and its functions."), (window_frame_offsets, WindowFrameOffsets, POST_MIGRATION_READER_VERSION, Semantic, "`ROWS` window frame bounds at an offset from the current row (`n PRECEDING`, `n FOLLOWING`)."), @@ -166,6 +168,7 @@ pub fn feature_of(construct: Construct<'_>) -> Option { match construct { Construct::Node(node) => match node { PhysNode::Limit { offset, .. } => (*offset != 0).then_some(Feature::LimitOffset), + PhysNode::HashJoin(j) => j.single_match.then_some(Feature::JoinSingleMatch), PhysNode::Window { .. } => Some(Feature::Window), PhysNode::Expand { .. } => Some(Feature::Expand), PhysNode::WindowGroupLimit { .. } => Some(Feature::WindowGroupLimit), @@ -177,7 +180,6 @@ pub fn feature_of(construct: Construct<'_>) -> Option { | PhysNode::Sort { .. } | PhysNode::TopN { .. } | PhysNode::Aggregate { .. } - | PhysNode::HashJoin(_) | PhysNode::NestedLoopJoin(_) | PhysNode::Union { .. } | PhysNode::WriteFiles { .. } @@ -193,6 +195,7 @@ pub fn feature_of(construct: Construct<'_>) -> Option { | JoinType::LeftSemi | JoinType::LeftAnti | JoinType::NullAwareAnti => None, + JoinType::Existence => Some(Feature::ExistenceJoin), JoinType::RightSemi | JoinType::RightAnti | JoinType::RightOuter => { Some(Feature::JoinBuildPreserved) } diff --git a/crates/krishiv-plan-codec/src/proto/krishiv.plan.v1.pb.rs b/crates/krishiv-plan-codec/src/proto/krishiv.plan.v1.pb.rs index 255d188d3..2429f121c 100644 --- a/crates/krishiv-plan-codec/src/proto/krishiv.plan.v1.pb.rs +++ b/crates/krishiv-plan-codec/src/proto/krishiv.plan.v1.pb.rs @@ -463,6 +463,17 @@ pub struct HashJoinNode { /// `JOIN_DISTRIBUTION_UNSPECIFIED`, which decodes as broadcast. #[prost(enumeration = "JoinDistribution", tag = "9")] pub distribution: i32, + /// A left outer join that fails when a probe row matches more than once + /// (a correlated scalar subquery without an aggregate); only `LeftOuter` + /// joins set it. + /// feature: join_single_match + #[prost(bool, tag = "10")] + pub single_match: bool, + /// The build may keep each key once: only on a left semi, left anti or + /// existence join without a residual. + /// ignorable: a reader without it builds every row, with the same result. + #[prost(bool, tag = "11")] + pub distinct_build: bool, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct NestedLoopJoinNode { @@ -1184,6 +1195,10 @@ pub enum JoinType { RightAnti = 7, /// feature: join_build_preserved RightOuter = 8, + /// Spark's `ExistenceJoin`: every probe row once, plus a non-nullable + /// boolean `exists` (some build row matched). + /// feature: existence_join + Existence = 9, } impl JoinType { /// String value of the enum field names used in the ProtoBuf definition. @@ -1201,6 +1216,7 @@ impl JoinType { Self::RightSemi => "JOIN_TYPE_RIGHT_SEMI", Self::RightAnti => "JOIN_TYPE_RIGHT_ANTI", Self::RightOuter => "JOIN_TYPE_RIGHT_OUTER", + Self::Existence => "JOIN_TYPE_EXISTENCE", } } /// Creates an enum from field names used in the ProtoBuf definition. @@ -1215,6 +1231,7 @@ impl JoinType { "JOIN_TYPE_RIGHT_SEMI" => Some(Self::RightSemi), "JOIN_TYPE_RIGHT_ANTI" => Some(Self::RightAnti), "JOIN_TYPE_RIGHT_OUTER" => Some(Self::RightOuter), + "JOIN_TYPE_EXISTENCE" => Some(Self::Existence), _ => None, } } diff --git a/crates/krishiv-plan-codec/tests/common/inventory.rs b/crates/krishiv-plan-codec/tests/common/inventory.rs index d76cc9adc..0953307f9 100644 --- a/crates/krishiv-plan-codec/tests/common/inventory.rs +++ b/crates/krishiv-plan-codec/tests/common/inventory.rs @@ -369,6 +369,8 @@ fn join(join_type: JoinType) -> HashJoinPlan { build_keys: vec![1, 0], null_safe: vec![false, true], residual: Some(residual()), + single_match: false, + distinct_build: false, runtime_filters: vec![JoinFilterPlan { id: FilterId(3), key: 0, @@ -386,6 +388,25 @@ fn join(join_type: JoinType) -> HashJoinPlan { } } +/// An existence or single-match join over one key, without a residual +/// condition or runtime filters (the single-match flag rides on a plain +/// left outer join). +fn subquery_join(join_type: JoinType) -> HashJoinPlan { + HashJoinPlan { + probe: boxed(scan()), + build: boxed(PhysNode::Scan(file_scan())), + join_type, + probe_keys: vec![0], + build_keys: vec![1], + null_safe: vec![false], + residual: None, + single_match: false, + distinct_build: false, + runtime_filters: Vec::new(), + distribution: JoinDistribution::Broadcast, + } +} + /// `ROLLUP (a, b)` over a two-column input: the key columns are kept or /// nulled and the last column is the grouping id. fn expand_rollup() -> PhysNode { @@ -652,6 +673,29 @@ pub(crate) fn nodes() -> Vec<(&'static str, PhysNode)> { ..join(JoinType::Inner) }), ), + ( + "hash_join_existence", + PhysNode::HashJoin(HashJoinPlan { + distinct_build: true, + ..subquery_join(JoinType::Existence) + }), + ), + ( + "hash_join_single_match", + PhysNode::HashJoin(HashJoinPlan { + single_match: true, + ..subquery_join(JoinType::LeftOuter) + }), + ), + ( + "nested_loop_existence", + PhysNode::NestedLoopJoin(NestedLoopPlan { + probe: boxed(scan()), + build: boxed(PhysNode::OneRow), + join_type: JoinType::Existence, + residual: Some(residual()), + }), + ), ( "nested_loop_join", PhysNode::NestedLoopJoin(NestedLoopPlan { diff --git a/crates/krishiv-plan-codec/tests/features.rs b/crates/krishiv-plan-codec/tests/features.rs index 90cd1c973..224411c37 100644 --- a/crates/krishiv-plan-codec/tests/features.rs +++ b/crates/krishiv-plan-codec/tests/features.rs @@ -100,6 +100,40 @@ fn constructs_map_to_one_feature_or_the_baseline() { feature_of(Construct::Join(krishiv_exec::JoinType::LeftSemi)), None ); + assert_eq!( + feature_of(Construct::Join(krishiv_exec::JoinType::Existence)), + Some(Feature::ExistenceJoin) + ); + // The single-match flag is a property of the hash join node: a plain + // left outer join is the baseline, a single-match one needs the + // feature. `distinct_build` needs none: a reader that ignores it builds + // every row and returns the same result. + let join = |single_match, distinct_build, join_type| { + let (_, plan) = cases() + .into_iter() + .find(|(n, _)| n == "hash_join_single_match") + .unwrap(); + match plan.root() { + PhysNode::HashJoin(join) => feature_of(Construct::Node(&PhysNode::HashJoin( + krishiv_exec::physical::HashJoinPlan { + single_match, + distinct_build, + join_type, + ..join.clone() + }, + ))), + _ => None, + } + }; + let (outer, semi) = ( + krishiv_exec::JoinType::LeftOuter, + krishiv_exec::JoinType::LeftSemi, + ); + assert_eq!(join(false, false, outer), None); + assert_eq!(join(true, false, outer), Some(Feature::JoinSingleMatch)); + assert_eq!(join(false, true, semi), None); + assert_eq!(Feature::ExistenceJoin.kind(), FeatureKind::Semantic); + assert_eq!(Feature::JoinSingleMatch.kind(), FeatureKind::Semantic); assert_eq!( feature_of(Construct::Partitioning( &krishiv_exec::physical::Partitioning::Single diff --git a/crates/krishiv-plan-codec/tests/golden/existence_join/hash_join_existence.bin b/crates/krishiv-plan-codec/tests/golden/existence_join/hash_join_existence.bin new file mode 100644 index 000000000..453fcc40a Binary files /dev/null and b/crates/krishiv-plan-codec/tests/golden/existence_join/hash_join_existence.bin differ diff --git a/crates/krishiv-plan-codec/tests/golden/existence_join/nested_loop_existence.bin b/crates/krishiv-plan-codec/tests/golden/existence_join/nested_loop_existence.bin new file mode 100644 index 000000000..64930f3b3 Binary files /dev/null and b/crates/krishiv-plan-codec/tests/golden/existence_join/nested_loop_existence.bin differ diff --git a/crates/krishiv-plan-codec/tests/golden/join_single_match/hash_join_single_match.bin b/crates/krishiv-plan-codec/tests/golden/join_single_match/hash_join_single_match.bin new file mode 100644 index 000000000..bb85a215f Binary files /dev/null and b/crates/krishiv-plan-codec/tests/golden/join_single_match/hash_join_single_match.bin differ diff --git a/crates/krishiv-plan-codec/tests/roundtrip.rs b/crates/krishiv-plan-codec/tests/roundtrip.rs index 11acb46db..82cfd9a6e 100644 --- a/crates/krishiv-plan-codec/tests/roundtrip.rs +++ b/crates/krishiv-plan-codec/tests/roundtrip.rs @@ -295,6 +295,9 @@ fn readers_that_predate_features_still_refuse_what_they_cannot_read() { // A feature added after the freeze: above every frozen version. "window_group_limit" => 6, n if n.starts_with("hash_join_right") => POST_MIGRATION_READER_VERSION, + "hash_join_existence" | "hash_join_single_match" | "nested_loop_existence" => { + POST_MIGRATION_READER_VERSION + } "window_frames" | "window_offset_values" => POST_MIGRATION_READER_VERSION, _ => MIN_READER_VERSION, }; @@ -469,6 +472,85 @@ fn joins_that_build_their_preserved_side_are_a_feature() { } } +/// The existence join type and the single-match flag are two semantic +/// features (no equivalent plan exists without either), each refused by +/// name by a reader without it; `distinct_build` needs none. +#[test] +fn existence_joins_and_single_match_are_features() { + for (name, feature) in [ + ("hash_join_existence", Feature::ExistenceJoin), + ("nested_loop_existence", Feature::ExistenceJoin), + ("hash_join_single_match", Feature::JoinSingleMatch), + ] { + assert_eq!(feature.kind(), FeatureKind::Semantic); + let m = message(name); + assert_eq!(m.features, [feature.id()], "{name}"); + assert_eq!( + m.min_reader_version, POST_MIGRATION_READER_VERSION, + "{name}" + ); + // A reader without the feature refuses the plan, naming it. + let (_, plan) = cases().into_iter().find(|(n, _)| n == name).unwrap(); + let bytes = encode(&plan).unwrap(); + let reader = ReaderFeatures::only([Feature::LimitOffset]); + let err = decode_with(&bytes, &Limits::default(), &reader).unwrap_err(); + assert_eq!( + err, + CodecError::UnsupportedFeatures { + missing: vec![feature.id().to_owned()] + }, + "{name}" + ); + // A reader with every feature reads it back exactly. + assert_eq!(round_trip(&plan), plan, "{name}"); + } + assert_eq!(Feature::ExistenceJoin.to_string(), "existence_join"); + assert_eq!(Feature::JoinSingleMatch.to_string(), "join_single_match"); +} + +/// `distinct_build` is only legal where the build's rows are never output +/// nor read beyond their keys; anywhere else the plan is corrupt. +#[test] +fn a_misplaced_distinct_build_flag_is_corrupt() { + use pb::node::Kind; + // `hash_join` is an inner join with a residual condition. + let mut m = message("hash_join"); + let last = m.nodes.len() - 1; + let Kind::HashJoin(j) = node_mut(&mut m, last) else { + panic!("not a join") + }; + j.distinct_build = true; + corrupt( + &m, + "a distinct build on a join that outputs or reads build rows", + ); + // An existence join with a residual condition reads build columns. + let mut m = message("hash_join"); + let Kind::HashJoin(j) = node_mut(&mut m, last) else { + panic!("not a join") + }; + j.join_type = pb::JoinType::Existence as i32; + j.distinct_build = true; + m.features = vec!["existence_join".to_owned()]; + corrupt( + &m, + "a distinct build on a join that outputs or reads build rows", + ); +} + +/// A single match on a join that is not left outer is corrupt. +#[test] +fn a_misplaced_single_match_flag_is_corrupt() { + use pb::node::Kind; + let mut m = message("hash_join"); + let last = m.nodes.len() - 1; + let Kind::HashJoin(j) = node_mut(&mut m, last) else { + panic!("not a join") + }; + j.single_match = true; + corrupt(&m, "single match"); +} + #[test] fn an_unspecified_join_type_is_corrupt() { let mut m = message("hash_join"); diff --git a/proto/krishiv/plan/v1/plan.proto b/proto/krishiv/plan/v1/plan.proto index d91e3cbd9..d85ea380f 100644 --- a/proto/krishiv/plan/v1/plan.proto +++ b/proto/krishiv/plan/v1/plan.proto @@ -375,6 +375,10 @@ enum JoinType { JOIN_TYPE_RIGHT_ANTI = 7; // feature: join_build_preserved JOIN_TYPE_RIGHT_OUTER = 8; + // Spark's `ExistenceJoin`: every probe row once, plus a non-nullable + // boolean `exists` (some build row matched). + // feature: existence_join + JOIN_TYPE_EXISTENCE = 9; } message Residual { @@ -394,6 +398,15 @@ message HashJoinNode { // v2: how matching rows meet in one task. v1 bytes leave this as // `JOIN_DISTRIBUTION_UNSPECIFIED`, which decodes as broadcast. JoinDistribution distribution = 9; + // A left outer join that fails when a probe row matches more than once + // (a correlated scalar subquery without an aggregate); only `LeftOuter` + // joins set it. + // feature: join_single_match + bool single_match = 10; + // The build may keep each key once: only on a left semi, left anti or + // existence join without a residual. + // ignorable: a reader without it builds every row, with the same result. + bool distinct_build = 11; } enum JoinDistribution {