diff --git a/crates/krishiv-exec/src/join/mod.rs b/crates/krishiv-exec/src/join/mod.rs index a5206aa69..79e3396a0 100644 --- a/crates/krishiv-exec/src/join/mod.rs +++ b/crates/krishiv-exec/src/join/mod.rs @@ -152,8 +152,10 @@ use crate::operators::Predicate; use crate::pipeline::{ExecNode, PipelineBuilder}; use crate::rows; -/// The join type. The probe (left) side is preserved by the outer and anti -/// joins; the build (right) side never is. +/// The join type. The probe (left) side is preserved by the left outer and +/// anti joins; the build side is preserved by the right semi, right anti +/// and right outer joins, which build the input whose rows they return +/// (design P2-T166). #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum JoinType { /// Matching pairs. @@ -166,12 +168,30 @@ pub enum JoinType { LeftAnti, /// `NOT IN`: see the module docs. NullAwareAnti, + /// 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. + RightSemi, + /// Build rows with no match: a left anti join that builds its left + /// input. Planned data only for now. + RightAnti, + /// Matching pairs, and each unmatched build row with NULL probe + /// columns: a left outer join that builds its preserved input. Planned + /// data only for now. + RightOuter, } impl JoinType { fn emits_build(self) -> bool { matches!(self, Self::Inner | Self::LeftOuter) } + + /// Whether the join returns rows of its build side by their matches + /// (right semi, right anti, right outer). + #[must_use] + pub fn builds_preserved(self) -> bool { + matches!(self, Self::RightSemi | Self::RightAnti | Self::RightOuter) + } } /// What a hash join joins. @@ -403,6 +423,11 @@ impl Shared { "a hash join needs matching, non-empty key lists".to_owned(), )); } + if join_type.builds_preserved() { + return Err(unsupported( + "hash joins that return rows of their build side by their matches", + )); + } if join_type == JoinType::NullAwareAnti && (n != 1 || filter.is_some() || null_safe.contains(&true)) { @@ -759,6 +784,40 @@ mod tests { } } + /// The joins that build their preserved side are plan data only: no + /// hash join runs them yet, and none is planned. + #[test] + fn joins_that_build_their_preserved_side_are_refused() { + for t in [ + JoinType::RightSemi, + JoinType::RightAnti, + JoinType::RightOuter, + ] { + assert!(t.builds_preserved()); + let e = HashJoin::try_new(spec(t, 1, false)) + .err() + .unwrap_or_else(|| panic!("{t:?} was accepted")); + assert_eq!( + e.error_class(), + krishiv_common::error::ErrorClass::KrishivUnsupportedFeature, + "{t:?}" + ); + assert!( + e.to_string().contains("return rows of their build side"), + "{e}" + ); + } + for t in [ + JoinType::Inner, + JoinType::LeftOuter, + JoinType::LeftSemi, + JoinType::LeftAnti, + JoinType::NullAwareAnti, + ] { + assert!(!t.builds_preserved(), "{t:?}"); + } + } + #[test] fn specs_are_validated() { assert!(HashJoin::try_new(spec(JoinType::Inner, 0, false)).is_err()); diff --git a/crates/krishiv-exec/src/join/nested.rs b/crates/krishiv-exec/src/join/nested.rs index dfb58efb1..3f2052b7e 100644 --- a/crates/krishiv-exec/src/join/nested.rs +++ b/crates/krishiv-exec/src/join/nested.rs @@ -67,11 +67,17 @@ impl NestedLoopJoin { /// /// # Errors /// `KRISHIV_UNSUPPORTED_FEATURE` for the null-aware anti join (it needs - /// an equality key; the planner uses the hash join). + /// an equality key; the planner uses the hash join) and for the joins + /// that return rows of their build side by their matches. pub fn try_new(spec: NestedLoopSpec) -> Result { if spec.join_type == JoinType::NullAwareAnti { return Err(unsupported("a null-aware anti join without a key")); } + if spec.join_type.builds_preserved() { + return Err(unsupported( + "nested-loop joins that return rows of their build side by their matches", + )); + } let fields = |nullable: bool| -> Vec { spec.probe_schema .fields() diff --git a/crates/krishiv-exec/src/join/probe.rs b/crates/krishiv-exec/src/join/probe.rs index 282003af3..8dca8d045 100644 --- a/crates/krishiv-exec/src/join/probe.rs +++ b/crates/krishiv-exec/src/join/probe.rs @@ -266,7 +266,12 @@ impl Cursor { /// The rows the join type emits once every match is known. fn tail(&self, shared: &Shared) -> Result> { let want = match shared.join_type { - JoinType::Inner => return Ok(None), + // No probe row waits for the end: inner joins emit pairs as + // they match, and the joins that build their preserved side + // return build rows, never unmatched probe rows. + JoinType::Inner | JoinType::RightSemi | JoinType::RightAnti | JoinType::RightOuter => { + return Ok(None); + } JoinType::LeftSemi => MATCHED, JoinType::LeftOuter | JoinType::LeftAnti | JoinType::NullAwareAnti => UNMATCHED, }; diff --git a/crates/krishiv-exec/src/planner/build.rs b/crates/krishiv-exec/src/planner/build.rs index 1925f8ee3..eedd661ed 100644 --- a/crates/krishiv-exec/src/planner/build.rs +++ b/crates/krishiv-exec/src/planner/build.rs @@ -665,6 +665,26 @@ impl<'a> Planner<'a> { krishiv_optimizer::estimated_rows(plan, &stats) } + /// Whether every table `plan` reads reports statistics, so its estimate + /// does not rest on default row counts. A `WITH` reference counts as + /// default-sized, as the estimator treats it. + pub(crate) fn has_statistics(&self, plan: &Arc) -> bool { + use krishiv_optimizer::Statistics as _; + let stats = super::ProviderStatistics(Arc::clone(&self.cx.tables)); + let mut stack = vec![plan]; + while let Some(p) = stack.pop() { + match &p.node { + PlanNode::Scan(scan) if stats.table(scan).is_none() => return false, + // The estimator reads a reference to a `WITH` definition + // as a default-sized relation, not through its definition. + PlanNode::CteRef(_) => return false, + _ => {} + } + p.for_each_child(&mut |c| stack.push(c)); + } + true + } + fn scan( &mut self, plan: &Arc, diff --git a/crates/krishiv-exec/src/planner/join.rs b/crates/krishiv-exec/src/planner/join.rs index ba9a626eb..8982d6808 100644 --- a/crates/krishiv-exec/src/planner/join.rs +++ b/crates/krishiv-exec/src/planner/join.rs @@ -31,6 +31,7 @@ use krishiv_plan::{ use super::build::{Planned, Planner, unsupported}; use super::explain::ExplainNode; +use super::orient::{InputStats, StatsSource, orient}; use super::properties::Distribution; use crate::error::Result; use crate::join::JoinType; @@ -131,8 +132,23 @@ impl Planner<'_> { let build_exprs: Vec> = keys.iter().map(|k| Arc::clone(&k.1)).collect(); let (probe, probe_keys) = self.with_columns(probe, &probe_exprs)?; let (build, build_keys) = self.with_columns(build, &build_exprs)?; - let build_plan = if build_left { &logical.0 } else { &logical.1 }; + let (probe_plan, build_plan) = if build_left { + (&logical.1, &logical.0) + } else { + (&logical.0, &logical.1) + }; let build_rows = self.estimated_rows(build_plan); + // 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); + (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. @@ -200,6 +216,13 @@ impl Planner<'_> { for p in published { detail.push_str(&format!(", {p}")); } + if let Some((other, probe_stats, build_stats)) = preferred { + detail.push_str(&format!( + ", prefers {other:?} (probe: {}; build: {})", + probe_stats.describe(), + build_stats.describe() + )); + } let (probe_schema, build_schema) = (Arc::clone(&probe.schema), Arc::clone(&build.schema)); let plan = HashJoinPlan { probe: Box::new(probe.node), @@ -245,6 +268,21 @@ fn residual_of(probe: &Planned, build: &Planned, residual: &[&Expr]) -> Option { + /// The statistics of a join input planned as `schema` from `plan`. + fn input_stats(&self, plan: &Arc, schema: &arrow_schema::Schema) -> InputStats { + let rows = self.estimated_rows(plan); + let source = if self.has_statistics(plan) { + StatsSource::Statistics + } else { + StatsSource::Default + }; + InputStats { + rows, + bytes: estimated_bytes(rows, schema), + source, + } + } + /// A join without an equality conjunct: a nested-loop join building the /// right input. fn nested_loop( diff --git a/crates/krishiv-exec/src/planner/mod.rs b/crates/krishiv-exec/src/planner/mod.rs index e87a9ab2d..807e590f3 100644 --- a/crates/krishiv-exec/src/planner/mod.rs +++ b/crates/krishiv-exec/src/planner/mod.rs @@ -32,6 +32,7 @@ mod explain; mod file_table; mod join; pub(crate) mod nodes; +mod orient; mod properties; pub(crate) mod pushdown; mod runtime_filter; diff --git a/crates/krishiv-exec/src/planner/orient.rs b/crates/krishiv-exec/src/planner/orient.rs new file mode 100644 index 000000000..9ebce474e --- /dev/null +++ b/crates/krishiv-exec/src/planner/orient.rs @@ -0,0 +1,258 @@ +// SPDX-License-Identifier: Apache-2.0 +//! Which input of a hash join builds the table (design P2-T166). +//! +//! The planner builds the right input by default, as Spark's hash joins +//! do. A semi, anti or left outer join whose right input is much larger +//! than its left is better built on the left input, as a right semi, right +//! anti or right outer join: the probe marks the build rows it matches and +//! the join emits the build rows by their marks at the end. Those joins do +//! not run yet; until they do, the planner keeps the right build and +//! `EXPLAIN` shows the preference (`prefers RightSemi (probe: ...; build: +//! ...)`). The decision is +//! a pure function of the two inputs' statistics ([`InputStats`]), so the +//! same function serves plan-time estimates now and statistics merged from +//! materialized exchanges later; it never reads process state, so every +//! task of a stage that calls it with the same statistics agrees. +//! +//! - A null-aware anti join (`NOT IN`) is never reoriented: it reads +//! whether the whole subquery side holds a NULL key. +//! - 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. +//! - Otherwise the left input is built when the right one costs more than +//! [`STATS_FACTOR`] times as much (both sides estimated from table +//! statistics), or [`DEFAULT_FACTOR`] times (either side from default +//! estimates). A tie keeps the right input. + +use crate::join::JoinType; + +use super::join::BROADCAST_MAX_BYTES; + +/// How much larger the right input's table must be to build the left one, +/// when both estimates rest on table statistics. +pub(crate) const STATS_FACTOR: f64 = 2.0; + +/// The same factor when either estimate rests on defaults. +pub(crate) const DEFAULT_FACTOR: f64 = 8.0; + +/// Where an input's estimate comes from. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum StatsSource { + /// Some table below has no statistics: default row counts. + Default, + /// Every table below has statistics (row counts, column statistics). + Statistics, +} + +/// What is known of one join input. +#[derive(Debug, Clone, Copy, PartialEq)] +pub(crate) struct InputStats { + /// Estimated rows. + pub(crate) rows: f64, + /// Estimated bytes of the hash table built from this input. + pub(crate) bytes: f64, + /// Where the estimate comes from. + pub(crate) source: StatsSource, +} + +impl InputStats { + /// `rows / MiB, source` for `EXPLAIN`. + pub(crate) fn describe(&self) -> String { + let source = match self.source { + StatsSource::Default => "default", + StatsSource::Statistics => "statistics", + }; + format!( + "{:.0} rows / {:.1} MiB, {source}", + self.rows, + self.bytes / (1024.0 * 1024.0) + ) + } +} + +/// 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 { + match t { + JoinType::LeftSemi => Some(JoinType::RightSemi), + JoinType::LeftAnti => Some(JoinType::RightAnti), + JoinType::LeftOuter => Some(JoinType::RightOuter), + JoinType::Inner + | JoinType::NullAwareAnti + | JoinType::RightSemi + | JoinType::RightAnti + | JoinType::RightOuter => None, + } +} + +/// 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 { + return join_type; + }; + if right.bytes <= BROADCAST_MAX_BYTES { + return join_type; + } + let factor = + if left.source == StatsSource::Statistics && right.source == StatsSource::Statistics { + STATS_FACTOR + } else { + DEFAULT_FACTOR + }; + // NaN compares false: no flip. + if right.bytes > factor * left.bytes { + other + } else { + join_type + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const MIB: f64 = 1024.0 * 1024.0; + + fn stats(bytes: f64, source: StatsSource) -> InputStats { + InputStats { + rows: bytes / 32.0, + bytes, + source, + } + } + + fn known(bytes: f64) -> InputStats { + stats(bytes, StatsSource::Statistics) + } + + #[test] + fn a_much_larger_right_input_flips_semi_anti_and_left_outer_joins() { + let (small, large) = (known(20.0 * MIB), known(500.0 * MIB)); + for (t, want) in [ + (JoinType::LeftSemi, JoinType::RightSemi), + (JoinType::LeftAnti, JoinType::RightAnti), + (JoinType::LeftOuter, JoinType::RightOuter), + ] { + assert_eq!(orient(t, &small, &large), want, "{t:?}"); + // The other way round keeps the right build. + assert_eq!(orient(t, &large, &small), t, "{t:?}"); + } + } + + #[test] + fn inner_null_aware_and_flipped_joins_keep_their_build() { + let (small, large) = (known(1.0 * MIB), known(5000.0 * MIB)); + for t in [ + JoinType::Inner, + JoinType::NullAwareAnti, + JoinType::RightSemi, + JoinType::RightAnti, + JoinType::RightOuter, + ] { + assert_eq!(orient(t, &small, &large), t, "{t:?}"); + assert_eq!(flipped(t), None, "{t:?}"); + } + } + + #[test] + fn a_broadcastable_right_input_stays_the_build() { + let left = known(1.0); + assert_eq!( + orient(JoinType::LeftSemi, &left, &known(BROADCAST_MAX_BYTES)), + JoinType::LeftSemi + ); + assert_eq!( + orient(JoinType::LeftSemi, &left, &known(BROADCAST_MAX_BYTES + 1.0)), + JoinType::RightSemi + ); + } + + #[test] + fn the_factor_depends_on_where_the_estimates_come_from() { + 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), + JoinType::LeftAnti + ); + assert_eq!( + orient(JoinType::LeftAnti, &known(49.0 * MIB), &right), + JoinType::RightAnti + ); + // A default estimate on either side needs a ratio above 8. + for (l, r) in [ + (StatsSource::Default, StatsSource::Statistics), + (StatsSource::Statistics, StatsSource::Default), + ] { + let right = stats(100.0 * MIB, r); + assert_eq!( + orient(JoinType::LeftSemi, &stats(20.0 * MIB, l), &right), + JoinType::LeftSemi + ); + assert_eq!( + orient(JoinType::LeftSemi, &stats(12.0 * MIB, l), &right), + JoinType::RightSemi + ); + } + } + + #[test] + fn unusable_estimates_keep_the_right_build() { + for bad in [f64::NAN, f64::INFINITY] { + let left = known(bad); + assert_eq!( + orient(JoinType::LeftSemi, &left, &known(100.0 * MIB)), + JoinType::LeftSemi + ); + } + assert_eq!( + orient(JoinType::LeftSemi, &known(1.0), &known(f64::NAN)), + JoinType::LeftSemi + ); + } + + /// The decision is a function of its inputs only: any order of calls + /// over the same inputs gives the same answers. + #[test] + fn orientation_is_deterministic() { + let sources = [StatsSource::Default, StatsSource::Statistics]; + let types = [ + JoinType::Inner, + JoinType::LeftOuter, + JoinType::LeftSemi, + JoinType::LeftAnti, + JoinType::NullAwareAnti, + ]; + let mut cases = Vec::new(); + let mut x: u64 = 0x9e37_79b9_7f4a_7c15; + for i in 0..512 { + // A fixed xorshift sequence of sizes from 1 B to ~1 TiB. + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + #[expect(clippy::cast_precision_loss, reason = "test sizes")] + let (a, b) = ((x % (1 << 40)) as f64, ((x >> 20) % (1 << 40)) as f64); + let t = types[i % types.len()]; + let l = stats(a, sources[i % 2]); + 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 reversed: Vec<_> = cases + .iter() + .rev() + .map(|(t, l, r)| orient(*t, l, r)) + .collect(); + assert!(first.iter().eq(reversed.iter().rev())); + assert!(first.iter().any(|t| matches!( + t, + JoinType::RightSemi | JoinType::RightAnti | JoinType::RightOuter + ))); + assert!(first.iter().any(|t| matches!( + t, + JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftOuter + ))); + } +} diff --git a/crates/krishiv-exec/tests/nested_loop.rs b/crates/krishiv-exec/tests/nested_loop.rs index 67788034c..43d7de1fe 100644 --- a/crates/krishiv-exec/tests/nested_loop.rs +++ b/crates/krishiv-exec/tests/nested_loop.rs @@ -198,6 +198,24 @@ fn null_aware_and_weighted_inputs_are_refused() { }) .unwrap_err(); assert_eq!(e.error_class(), ErrorClass::KrishivUnsupportedFeature); + for join_type in [ + JoinType::RightSemi, + JoinType::RightAnti, + JoinType::RightOuter, + ] { + let e = NestedLoopJoin::try_new(NestedLoopSpec { + join_type, + probe_schema: schema("p"), + build_schema: schema("b"), + filter: None, + }) + .unwrap_err(); + assert_eq!(e.error_class(), ErrorClass::KrishivUnsupportedFeature); + assert!( + e.to_string().contains("return rows of their build side"), + "{e}" + ); + } let weighted = |name: &str| { let b = &batches(name, &[Some(1), Some(2)], 2)[0]; vec![Batch::with_weights(b.data().clone(), Weights::Const(2)).unwrap()] diff --git a/crates/krishiv-exec/tests/planner/main.rs b/crates/krishiv-exec/tests/planner/main.rs index 74c038df8..fc0db9064 100644 --- a/crates/krishiv-exec/tests/planner/main.rs +++ b/crates/krishiv-exec/tests/planner/main.rs @@ -16,6 +16,7 @@ mod folding; mod group_limit; mod hints; mod listing; +mod orientation; mod physical; mod runtime_filters; mod semantics; diff --git a/crates/krishiv-exec/tests/planner/orientation.rs b/crates/krishiv-exec/tests/planner/orientation.rs new file mode 100644 index 000000000..04f77a2fa --- /dev/null +++ b/crates/krishiv-exec/tests/planner/orientation.rs @@ -0,0 +1,164 @@ +// SPDX-License-Identifier: Apache-2.0 +//! Which input a hash join prefers to build (design P2-T166): a semi, anti +//! or left outer join whose right input is much larger than its left +//! prefers to build the left one. Those joins do not run yet, so the plan +//! keeps the right build and `EXPLAIN` shows the preference. + +use std::collections::HashMap; +use std::sync::Arc; + +use krishiv_exec::planner::{MemoryTable, MemoryTables}; +use krishiv_optimizer::TableStats; +use krishiv_plan::analyzer::testing::context; +use krishiv_plan::analyzer::{MemoryCatalog, ResolutionConf, TableColumn}; + +use super::{Env, table}; + +/// `t` (10 rows), `big` and `mid` (the same rows, statistics claiming 100 +/// and 5 million), and `bare` (no statistics: a default estimate of one +/// million rows). +fn env() -> Env { + let (schema, batches) = table(10); + let columns: Vec = schema + .fields() + .iter() + .map(|f| TableColumn { + name: f.name().clone(), + data_type: krishiv_types::DataType::from_arrow(f.data_type()).unwrap(), + nullable: f.is_nullable(), + }) + .collect(); + let mut catalog = MemoryCatalog::new("spark_catalog"); + let mut tables = MemoryTables::new(); + for (name, rows) in [ + ("t", Some(10)), + ("big", Some(100_000_000)), + ("mid", Some(5_000_000)), + ("bare", None), + ] { + catalog = catalog.with_table(&["default"], name, columns.clone(), None); + let stats = rows.map(|rows| TableStats { + rows, + columns: HashMap::new(), + }); + let table = MemoryTable::new(Arc::clone(&schema), batches.clone()).with_statistics(stats); + tables = tables.with_table(&["spark_catalog", "default", name], Arc::new(table)); + } + Env { + ctx: context(catalog, ResolutionConf::default()), + tables: Arc::new(tables), + } +} + +/// The `HashJoin` lines of the plan of `sql`. +fn joins(env: &Env, sql: &str) -> Vec { + let plan = env.plan(sql).unwrap_or_else(|e| panic!("{sql}: {e}")); + plan.to_string() + .lines() + .filter_map(|l| l.find("HashJoin").map(|i| l[i..].to_owned())) + .collect() +} + +#[test] +fn a_much_larger_right_input_is_shown_as_the_side_not_to_build() { + let env = env(); + let cases = [ + ( + "SELECT a FROM t WHERE EXISTS (SELECT 1 FROM big WHERE big.a = t.a)", + "HashJoin LeftSemi, [a#0] = [a#3], RuntimeFilter #1: a#0 -> spark_catalog.default.t.a, \ + prefers RightSemi (probe: 10 rows / 0.0 MiB, statistics; build: 100000000 rows / 1144.4 MiB, statistics)", + ), + ( + "SELECT a FROM t WHERE NOT EXISTS (SELECT 1 FROM big WHERE big.a = t.a)", + "HashJoin LeftAnti, [a#0] = [a#3], \ + prefers RightAnti (probe: 10 rows / 0.0 MiB, statistics; build: 100000000 rows / 1144.4 MiB, statistics)", + ), + ]; + for (sql, want) in cases { + assert_eq!(joins(&env, sql), [want], "{sql}"); + } + let outer = joins( + &env, + "SELECT t.a, big.b FROM t LEFT JOIN big ON t.a = big.a", + ); + assert_eq!(outer.len(), 1); + assert!(outer[0].starts_with("HashJoin LeftOuter"), "{outer:?}"); + assert!( + outer[0].contains(", prefers RightOuter (probe: 10 rows"), + "{outer:?}" + ); +} + +#[test] +fn null_aware_inner_small_and_hinted_joins_show_no_preference() { + let env = env(); + for (sql, want) in [ + // `NOT IN` reads whether the whole subquery side holds a NULL. + ( + "SELECT a FROM t WHERE a NOT IN (SELECT a FROM big)", + "HashJoin NullAwareAnti, [a#0] = [a#3]", + ), + // Inner joins keep the join orderer's choice. + ( + "SELECT t.a FROM t JOIN big ON t.a = big.a", + "HashJoin Inner, [a#0] = [a#3], RuntimeFilter #1: a#0 -> spark_catalog.default.t.a", + ), + // A build small enough to broadcast stays. + ( + "SELECT a FROM big WHERE EXISTS (SELECT 1 FROM t WHERE big.a = t.a)", + "HashJoin LeftSemi, [a#0] = [a#3], \ + RuntimeFilter #1: a#0 -> spark_catalog.default.big.a (bloom)", + ), + // A hint decides. + ( + "SELECT /*+ SHUFFLE_HASH(big) */ t.a FROM t LEFT JOIN big ON t.a = big.a", + "HashJoin LeftOuter, [a#0] = [a#3]", + ), + ] { + assert_eq!(joins(&env, sql), [want], "{sql}"); + } +} + +#[test] +fn a_default_estimate_needs_a_larger_ratio() { + let env = env(); + // 5M rows against a default 1M: 5x is enough with statistics on both + // sides (above 2x), not with a default estimate on one (8x). + let sql = "SELECT a FROM bare WHERE EXISTS (SELECT 1 FROM mid WHERE mid.a = bare.a)"; + // No preference: `bare` probes a `LeftSemi` that builds `mid`. + assert_eq!( + joins(&env, sql), + ["HashJoin LeftSemi, [a#0] = [a#3], RuntimeFilter #1: a#0 -> spark_catalog.default.bare.a"], + "{sql}" + ); + let sql = "SELECT a FROM bare WHERE EXISTS (SELECT 1 FROM big WHERE big.a = bare.a)"; + let js = joins(&env, sql); + assert!( + js.iter().any(|j| j.contains( + "prefers RightSemi (probe: 1000000 rows / 11.4 MiB, default; build: 100000000" + )), + "{js:?}" + ); + let sql = "SELECT a FROM t WHERE EXISTS (SELECT 1 FROM mid WHERE mid.a = t.a)"; + assert!( + joins(&env, sql) + .iter() + .any(|j| j.contains("prefers RightSemi")), + "{sql}" + ); +} + +/// A `WITH` reference reads as a default-sized relation (the estimator +/// does not look through it), so it is not taken as resting on statistics. +#[test] +fn a_cte_reference_is_a_default_estimate() { + let env = env(); + let sql = "WITH c AS (SELECT a FROM big) \ + SELECT a FROM t WHERE EXISTS (SELECT 1 FROM c WHERE c.a = t.a)"; + let js = joins(&env, sql); + assert_eq!(js.len(), 1, "{js:?}"); + assert!( + js[0].ends_with("prefers RightSemi (probe: 10 rows / 0.0 MiB, statistics; build: 1000000 rows / 11.4 MiB, default)"), + "{js:?}" + ); +} diff --git a/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q08.snap b/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q08.snap index f2d09d2ef..ea1bb47da 100644 --- a/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q08.snap +++ b/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q08.snap @@ -9,38 +9,38 @@ Sort [o_year#60 ASC NULLS FIRST] +- Exchange hashpartitioning(o_year#60) +- Project [o_year#60, CASE WHEN (nation#62 == 'BRAZIL') THEN volume#61 ELSE 0.0000 END, volume#61] +- Project [extract('year', o_orderdate#36) AS o_year#60, (l_extendedprice#21 * (1 - l_discount#22)) AS volume#61, n_name#54 AS nation#62] - +- HashJoin Inner, [n_nationkey#53] = [s_nationkey#12], RuntimeFilter #7: n_nationkey#53 -> spark_catalog.default.nation.n_nationkey - :- Project [n_nationkey#53, n_name#54] - : +- Scan spark_catalog.default.nation [n_nationkey#49, n_name#50, n_regionkey#51] (ids +4) + +- HashJoin Inner, [l_suppkey#18] = [s_suppkey#9], RuntimeFilter #7: l_suppkey#18 -> spark_catalog.default.lineitem.l_suppkey + :- Project [l_suppkey#18, l_extendedprice#21, l_discount#22, o_orderdate#36] + : +- HashJoin Inner, [l_partkey#17] = [p_partkey#0], RuntimeFilter #5: l_partkey#17 -> spark_catalog.default.lineitem.l_partkey + : :- Project [l_partkey#17, l_suppkey#18, l_extendedprice#21, l_discount#22, o_orderdate#36] + : : +- HashJoin Inner, [l_orderkey#16] = [o_orderkey#32], RuntimeFilter #4: l_orderkey#16 -> spark_catalog.default.lineitem.l_orderkey + : : :- Scan spark_catalog.default.lineitem [l_orderkey#16, l_partkey#17, l_suppkey#18, l_extendedprice#21, l_discount#22] + : : +- BroadcastExchange + : : +- Project [o_orderkey#32, o_orderdate#36] + : : +- HashJoin Inner, [o_custkey#33] = [c_custkey#41], RuntimeFilter #3: o_custkey#33 -> spark_catalog.default.orders.o_custkey + : : :- Filter ((o_orderdate#36 >= DATE '1995-01-01') and (o_orderdate#36 <= DATE '1996-12-31')) + : : : +- Scan spark_catalog.default.orders [o_orderkey#32, o_custkey#33, o_orderdate#36], PushedFilters: [((o_orderdate#36 >= DATE '1995-01-01') and (o_orderdate#36 <= DATE '1996-12-31'))] + : : +- BroadcastExchange + : : +- Project [c_custkey#41] + : : +- HashJoin Inner, [c_nationkey#44] = [n_nationkey#49], RuntimeFilter #2: c_nationkey#44 -> spark_catalog.default.customer.c_nationkey + : : :- Scan spark_catalog.default.customer [c_custkey#41, c_nationkey#44] + : : +- BroadcastExchange + : : +- Project [n_nationkey#49] + : : +- HashJoin Inner, [n_regionkey#51] = [r_regionkey#57], RuntimeFilter #1: n_regionkey#51 -> spark_catalog.default.nation.n_regionkey + : : :- Project [n_nationkey#49, n_regionkey#51] + : : : +- Scan spark_catalog.default.nation [n_nationkey#49, n_name#50, n_regionkey#51] + : : +- BroadcastExchange + : : +- Project [r_regionkey#57] + : : +- Filter (r_name#58 == 'AMERICA') + : : +- Scan spark_catalog.default.region [r_regionkey#57, r_name#58], PushedFilters: [(r_name#58 == 'AMERICA')] + : +- BroadcastExchange + : +- Project [p_partkey#0] + : +- Filter (p_type#4 == 'ECONOMY ANODIZED STEEL') + : +- Scan spark_catalog.default.part [p_partkey#0, p_type#4], PushedFilters: [(p_type#4 == 'ECONOMY ANODIZED STEEL')] +- BroadcastExchange - +- Project [s_nationkey#12, o_orderdate#36, l_extendedprice#21, l_discount#22] - +- HashJoin Inner, [s_suppkey#9] = [l_suppkey#18], RuntimeFilter #6: s_suppkey#9 -> spark_catalog.default.supplier.s_suppkey + +- Project [s_suppkey#9, n_name#54] + +- HashJoin Inner, [s_nationkey#12] = [n_nationkey#53], RuntimeFilter #6: s_nationkey#12 -> spark_catalog.default.supplier.s_nationkey :- Scan spark_catalog.default.supplier [s_suppkey#9, s_nationkey#12] +- BroadcastExchange - +- Project [o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - +- HashJoin Inner, [n_regionkey#51] = [r_regionkey#57], RuntimeFilter #5: n_regionkey#51 -> spark_catalog.default.nation.n_regionkey - :- Project [n_regionkey#51, o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- HashJoin Inner, [n_nationkey#49] = [c_nationkey#44], RuntimeFilter #4: n_nationkey#49 -> spark_catalog.default.nation.n_nationkey - : :- Project [n_nationkey#49, n_regionkey#51] - : : +- Scan spark_catalog.default.nation [n_nationkey#49, n_name#50, n_regionkey#51] - : +- BroadcastExchange - : +- Project [c_nationkey#44, o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- HashJoin Inner, [c_custkey#41] = [o_custkey#33], RuntimeFilter #3: c_custkey#41 -> spark_catalog.default.customer.c_custkey - : :- Scan spark_catalog.default.customer [c_custkey#41, c_nationkey#44] - : +- BroadcastExchange - : +- Project [o_custkey#33, o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- HashJoin Inner, [o_orderkey#32] = [l_orderkey#16], RuntimeFilter #2: o_orderkey#32 -> spark_catalog.default.orders.o_orderkey - : :- Filter ((o_orderdate#36 >= DATE '1995-01-01') and (o_orderdate#36 <= DATE '1996-12-31')) - : : +- Scan spark_catalog.default.orders [o_orderkey#32, o_custkey#33, o_orderdate#36], PushedFilters: [((o_orderdate#36 >= DATE '1995-01-01') and (o_orderdate#36 <= DATE '1996-12-31'))] - : +- BroadcastExchange - : +- Project [l_orderkey#16, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- HashJoin Inner, [l_partkey#17] = [p_partkey#0], RuntimeFilter #1: l_partkey#17 -> spark_catalog.default.lineitem.l_partkey - : :- Scan spark_catalog.default.lineitem [l_orderkey#16, l_partkey#17, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- BroadcastExchange - : +- Project [p_partkey#0] - : +- Filter (p_type#4 == 'ECONOMY ANODIZED STEEL') - : +- Scan spark_catalog.default.part [p_partkey#0, p_type#4], PushedFilters: [(p_type#4 == 'ECONOMY ANODIZED STEEL')] - +- BroadcastExchange - +- Project [r_regionkey#57] - +- Filter (r_name#58 == 'AMERICA') - +- Scan spark_catalog.default.region [r_regionkey#57, r_name#58], PushedFilters: [(r_name#58 == 'AMERICA')] + +- Project [n_nationkey#53, n_name#54] + +- Scan spark_catalog.default.nation [n_nationkey#49, n_name#50, n_regionkey#51] (ids +4) diff --git a/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q10.snap b/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q10.snap index 620c03b42..9399c5c6c 100644 --- a/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q10.snap +++ b/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q10.snap @@ -9,20 +9,20 @@ TopN 20, [revenue#37 DESC NULLS LAST] +- HashAggregate keys=[c_custkey#0, c_name#1, c_acctbal#5, c_phone#4, n_name#34, c_address#2, c_comment#7], functions=[sum((l_extendedprice#22 * (1 - l_discount#23)))] +- Exchange hashpartitioning(c_custkey#0, c_name#1, c_acctbal#5, c_phone#4, n_name#34, c_address#2, c_comment#7) +- Project [c_custkey#0, c_name#1, c_acctbal#5, c_phone#4, n_name#34, c_address#2, c_comment#7, (l_extendedprice#22 * (1 - l_discount#23))] - +- Project [n_name#34, c_custkey#0, c_name#1, c_address#2, c_phone#4, c_acctbal#5, c_comment#7, l_extendedprice#22, l_discount#23] - +- HashJoin Inner, [n_nationkey#33] = [c_nationkey#3], RuntimeFilter #3: n_nationkey#33 -> spark_catalog.default.nation.n_nationkey - :- Scan spark_catalog.default.nation [n_nationkey#33, n_name#34] + +- Project [l_extendedprice#22, l_discount#23, c_custkey#0, c_name#1, c_address#2, c_phone#4, c_acctbal#5, c_comment#7, n_name#34] + +- HashJoin Inner, [o_custkey#9] = [c_custkey#0], RuntimeFilter #3: o_custkey#9 -> spark_catalog.default.orders.o_custkey + :- Project [l_extendedprice#22, l_discount#23, o_custkey#9] + : +- HashJoin Inner, [l_orderkey#17] = [o_orderkey#8], RuntimeFilter #1: l_orderkey#17 -> spark_catalog.default.lineitem.l_orderkey + : :- Project [l_orderkey#17, l_extendedprice#22, l_discount#23] + : : +- Filter (l_returnflag#25 == 'R') + : : +- Scan spark_catalog.default.lineitem [l_orderkey#17, l_extendedprice#22, l_discount#23, l_returnflag#25], PushedFilters: [(l_returnflag#25 == 'R')] + : +- BroadcastExchange + : +- Project [o_orderkey#8, o_custkey#9] + : +- Filter ((o_orderdate#12 >= DATE '1993-10-01') and (o_orderdate#12 < DATE '1994-01-01')) + : +- Scan spark_catalog.default.orders [o_orderkey#8, o_custkey#9, o_orderdate#12], PushedFilters: [((o_orderdate#12 >= DATE '1993-10-01') and (o_orderdate#12 < DATE '1994-01-01'))] +- BroadcastExchange - +- Project [c_custkey#0, c_name#1, c_address#2, c_nationkey#3, c_phone#4, c_acctbal#5, c_comment#7, l_extendedprice#22, l_discount#23] - +- HashJoin Inner, [c_custkey#0] = [o_custkey#9], RuntimeFilter #2: c_custkey#0 -> spark_catalog.default.customer.c_custkey + +- Project [c_custkey#0, c_name#1, c_address#2, c_phone#4, c_acctbal#5, c_comment#7, n_name#34] + +- HashJoin Inner, [c_nationkey#3] = [n_nationkey#33], RuntimeFilter #2: c_nationkey#3 -> spark_catalog.default.customer.c_nationkey :- Scan spark_catalog.default.customer [c_custkey#0, c_name#1, c_address#2, c_nationkey#3, c_phone#4, c_acctbal#5, c_comment#7] +- BroadcastExchange - +- Project [o_custkey#9, l_extendedprice#22, l_discount#23] - +- HashJoin Inner, [o_orderkey#8] = [l_orderkey#17], RuntimeFilter #1: o_orderkey#8 -> spark_catalog.default.orders.o_orderkey - :- Project [o_orderkey#8, o_custkey#9] - : +- Filter ((o_orderdate#12 >= DATE '1993-10-01') and (o_orderdate#12 < DATE '1994-01-01')) - : +- Scan spark_catalog.default.orders [o_orderkey#8, o_custkey#9, o_orderdate#12], PushedFilters: [((o_orderdate#12 >= DATE '1993-10-01') and (o_orderdate#12 < DATE '1994-01-01'))] - +- BroadcastExchange - +- Project [l_orderkey#17, l_extendedprice#22, l_discount#23] - +- Filter (l_returnflag#25 == 'R') - +- Scan spark_catalog.default.lineitem [l_orderkey#17, l_extendedprice#22, l_discount#23, l_returnflag#25], PushedFilters: [(l_returnflag#25 == 'R')] + +- Scan spark_catalog.default.nation [n_nationkey#33, n_name#34] diff --git a/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q21.snap b/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q21.snap index da33645dd..fd337ca37 100644 --- a/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q21.snap +++ b/crates/krishiv-exec/tests/planner/snapshots/planner__tpch__q21.snap @@ -11,25 +11,25 @@ TopN 100, [numwait#68 DESC NULLS LAST, s_name#1 ASC NULLS FIRST] +- Project [s_name#1] +- HashJoin LeftAnti, [l_orderkey#7] = [l_orderkey#52], residual: (NOT (l_suppkey#54 == l_suppkey#9)) :- HashJoin LeftSemi, [l_orderkey#7] = [l_orderkey#36], residual: (NOT (l_suppkey#38 == l_suppkey#9)), RuntimeFilter #4: l_orderkey#7 -> spark_catalog.default.lineitem.l_orderkey - : :- Project [s_name#1, l_orderkey#7, l_suppkey#9] - : : +- HashJoin Inner, [s_nationkey#3] = [n_nationkey#32], RuntimeFilter #3: s_nationkey#3 -> spark_catalog.default.supplier.s_nationkey - : : :- Project [s_name#1, s_nationkey#3, l_orderkey#7, l_suppkey#9] - : : : +- HashJoin Inner, [s_suppkey#0] = [l_suppkey#9], RuntimeFilter #2: s_suppkey#0 -> spark_catalog.default.supplier.s_suppkey - : : : :- Scan spark_catalog.default.supplier [s_suppkey#0, s_name#1, s_nationkey#3] + : :- Project [l_orderkey#7, l_suppkey#9, s_name#1] + : : +- HashJoin Inner, [l_suppkey#9] = [s_suppkey#0], RuntimeFilter #3: l_suppkey#9 -> spark_catalog.default.lineitem.l_suppkey + : : :- Project [l_orderkey#7, l_suppkey#9] + : : : +- HashJoin Inner, [l_orderkey#7] = [o_orderkey#23], RuntimeFilter #1: l_orderkey#7 -> spark_catalog.default.lineitem.l_orderkey + : : : :- Project [l_orderkey#7, l_suppkey#9] + : : : : +- Filter (l_receiptdate#19 > l_commitdate#18) + : : : : +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_commitdate#18, l_receiptdate#19], PushedFilters: [(l_receiptdate#19 > l_commitdate#18)] : : : +- BroadcastExchange - : : : +- Project [l_orderkey#7, l_suppkey#9] - : : : +- HashJoin Inner, [l_orderkey#7] = [o_orderkey#23], RuntimeFilter #1: l_orderkey#7 -> spark_catalog.default.lineitem.l_orderkey - : : : :- Project [l_orderkey#7, l_suppkey#9] - : : : : +- Filter (l_receiptdate#19 > l_commitdate#18) - : : : : +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_commitdate#18, l_receiptdate#19], PushedFilters: [(l_receiptdate#19 > l_commitdate#18)] - : : : +- BroadcastExchange - : : : +- Project [o_orderkey#23] - : : : +- Filter (o_orderstatus#25 == 'F') - : : : +- Scan spark_catalog.default.orders [o_orderkey#23, o_orderstatus#25], PushedFilters: [(o_orderstatus#25 == 'F')] + : : : +- Project [o_orderkey#23] + : : : +- Filter (o_orderstatus#25 == 'F') + : : : +- Scan spark_catalog.default.orders [o_orderkey#23, o_orderstatus#25], PushedFilters: [(o_orderstatus#25 == 'F')] : : +- BroadcastExchange - : : +- Project [n_nationkey#32] - : : +- Filter (n_name#33 == 'SAUDI ARABIA') - : : +- Scan spark_catalog.default.nation [n_nationkey#32, n_name#33], PushedFilters: [(n_name#33 == 'SAUDI ARABIA')] + : : +- Project [s_suppkey#0, s_name#1] + : : +- HashJoin Inner, [s_nationkey#3] = [n_nationkey#32], RuntimeFilter #2: s_nationkey#3 -> spark_catalog.default.supplier.s_nationkey + : : :- Scan spark_catalog.default.supplier [s_suppkey#0, s_name#1, s_nationkey#3] + : : +- BroadcastExchange + : : +- Project [n_nationkey#32] + : : +- Filter (n_name#33 == 'SAUDI ARABIA') + : : +- Scan spark_catalog.default.nation [n_nationkey#32, n_name#33], PushedFilters: [(n_name#33 == 'SAUDI ARABIA')] : +- BroadcastExchange : +- Project [l_orderkey#36, l_suppkey#38] : +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_commitdate#18, l_receiptdate#19] (ids +29) diff --git a/crates/krishiv-optimizer/src/expr.rs b/crates/krishiv-optimizer/src/expr.rs index 140cbbaf4..a27aaa749 100644 --- a/crates/krishiv-optimizer/src/expr.rs +++ b/crates/krishiv-optimizer/src/expr.rs @@ -235,6 +235,49 @@ pub(crate) fn implied(pred: &Arc, side: &Attrs) -> Option> { or_all(out) } +/// `pred` with the column equalities every one of its disjuncts holds +/// taken out as conjuncts of their own: `(a = b and x) or (a = b and y)` is +/// `a = b` and `x or y` (distributivity, which holds in three-valued logic +/// too). A plain column equality cannot raise and is deterministic, so +/// evaluating it once instead of in each disjunct changes nothing. Join +/// ordering needs it: an `or` over three inputs whose disjuncts all join +/// the same keys (TPC-DS q13) is otherwise one condition over all three, +/// and the inputs meet only in a cross product. +pub(crate) fn common_equalities(pred: &Arc) -> Vec> { + let parts = disjuncts(pred); + if parts.len() < 2 || has_subquery(pred) { + return vec![Arc::clone(pred)]; + } + let each: Vec>> = parts.iter().map(conjuncts).collect(); + let is_column = |e: &Expr| matches!(e.node, ExprNode::Resolved(ResolvedExpr::Column(_))); + let common: Vec> = each[0] + .iter() + .filter(|c| matches!(builtin(c), Some((Builtin::Eq, [l, r], _)) if is_column(l) && is_column(r))) + .filter(|c| each[1..].iter().all(|d| d.iter().any(|x| same(x, c)))) + .cloned() + .collect(); + if common.is_empty() { + return vec![Arc::clone(pred)]; + } + let mut rest = Vec::with_capacity(each.len()); + for d in &each { + let left: Vec> = d + .iter() + .filter(|x| !common.iter().any(|c| same(x, c))) + .cloned() + .collect(); + match and_all(left) { + Some(e) => rest.push(e), + // A disjunct of only common equalities: the `or` always holds + // once they do. + None => return common, + } + } + let mut out = common; + out.extend(or_all(rest)); + out +} + /// `a and b and ...`, left-deep; `None` for no predicates. pub(crate) fn and_all(preds: impl IntoIterator>) -> Option> { preds diff --git a/crates/krishiv-optimizer/src/join_order.rs b/crates/krishiv-optimizer/src/join_order.rs index 9cd27440b..104f4ba58 100644 --- a/crates/krishiv-optimizer/src/join_order.rs +++ b/crates/krishiv-optimizer/src/join_order.rs @@ -23,7 +23,8 @@ use krishiv_plan::{ }; use crate::expr::{ - Attrs, and_all, column, conjuncts, has_subquery, implied, is_deterministic, references, + Attrs, and_all, column, common_equalities, conjuncts, has_subquery, implied, is_deterministic, + references, }; use crate::stats::{Est, Estimator, Statistics, equi_columns, selectivity}; @@ -229,7 +230,16 @@ impl Reorder<'_> { return self.children(root); } let ests: Vec> = leaves.iter().map(|l| self.est.plan(l)).collect(); - let graph = Graph::new(&leaves, &ests, preds); + let mut graph = Graph::new(&leaves, &ests, preds.clone()); + if graph.binary_parts() > graph.parts(true) { + // Some inputs are joined only by a condition over three or more + // of them, so two of them would meet in a cross product. Take + // the column equalities every disjunct of such a condition holds + // as conditions of their own (TPC-DS q13). Only then: elsewhere + // they would add join keys the inputs do not need. + let preds = preds.iter().flat_map(common_equalities).collect(); + graph = Graph::new(&leaves, &ests, preds); + } if graph.parts(false) < graph.parts(true) { // Inputs related only through a conjunct with a subquery: as // written they meet in a join on it; re-planned they would meet @@ -400,7 +410,7 @@ impl Graph { .collect() }) .collect(); - let mut kept = Vec::new(); + let mut joins = Vec::new(); for (p, r) in preds.into_iter().zip(&refs) { let mut on: Vec = r .iter() @@ -418,6 +428,24 @@ impl Graph { })); continue; } + joins.push((p, on)); + } + // A filtered input keeps at most as many distinct values of a column + // as it keeps rows, as the estimate of a filter does: a join with a + // filtered input then counts the distinct keys that input keeps + // (containment), not its whole table's. Without it, a dimension + // filtered to a year (365 of 73,049 dates) joined to a fact table + // holding five years of dates was estimated 40x too small, which + // made the fact side look like the smaller input to build. + for (i, ids) in attrs.iter().enumerate() { + for id in ids { + if let Some(c) = all.cols.get_mut(id) { + c.ndv = c.ndv.min(rows[i]); + } + } + } + let mut kept = Vec::new(); + for (p, on) in joins { let sel = conditional(&p, selectivity(&p, &all), &on, &attrs, |i| &ests[order[i]]); let edge = !has_subquery(&p); kept.push(Pred { @@ -455,6 +483,23 @@ impl Graph { (0..self.n()).filter(|&x| find(&mut root, x) == x).count() } + /// The connected parts through join conditions over exactly two inputs. + fn binary_parts(&self) -> usize { + let mut root: Vec = (0..self.n()).collect(); + fn find(root: &mut [usize], mut x: usize) -> usize { + while root[x] != x { + root[x] = root[root[x]]; + x = root[x]; + } + x + } + for p in self.preds.iter().filter(|p| p.edge && p.on.len() == 2) { + let (a, b) = (find(&mut root, p.on[0]), find(&mut root, p.on[1])); + root[a] = b; + } + (0..self.n()).filter(|&x| find(&mut root, x) == x).count() + } + fn n(&self) -> usize { self.leaves.len() } diff --git a/crates/krishiv-optimizer/src/stats.rs b/crates/krishiv-optimizer/src/stats.rs index 2484702a4..4b846359f 100644 --- a/crates/krishiv-optimizer/src/stats.rs +++ b/crates/krishiv-optimizer/src/stats.rs @@ -2,7 +2,8 @@ //! Table statistics and cardinality estimates. //! //! Estimates follow the textbook independence assumptions: a predicate's -//! selectivity is `1 / NDV` for an equality with a literal, the covered +//! selectivity is `1 / NDV` for an equality with a literal (when the +//! column's distinct count is known, else [`DEFAULT_SELECTIVITY`]), the covered //! fraction of `[min, max]` for a range, `1 / max(NDV)` for an equi-join, //! and [`DEFAULT_SELECTIVITY`] when nothing better is known; conjunctions //! multiply and disjunctions add. Without statistics a table holds @@ -77,8 +78,11 @@ pub fn estimated_rows(plan: &Arc, stats: &dyn Statistics) -> f64 { /// What is known of one attribute. #[derive(Debug, Clone)] pub(crate) struct Col { - /// Distinct non-null values, at least 1. + /// Distinct non-null values, at least 1. Without statistics, every + /// value is taken as distinct (`ndv` = rows). pub(crate) ndv: f64, + /// Whether `ndv` comes from statistics rather than that assumption. + pub(crate) ndv_known: bool, /// The fraction of rows that are null. pub(crate) nulls: f64, /// Bounds, as numbers on one scale. @@ -114,6 +118,7 @@ impl Est { fn distinct(rows: f64) -> Col { Col { ndv: rows.max(1.0), + ndv_known: false, nulls: 0.0, range: None, } @@ -264,6 +269,9 @@ impl<'a> Estimator<'a> { let mut mixed = false; let mut literals: Vec<&Literal> = Vec::new(); let mut other = false; + // Whether every input column's count comes from + // statistics (literals are counted exactly). + let mut known = true; for proj in projections.iter() { let s = strip_alias(&proj[j]); match &s.node { @@ -281,6 +289,7 @@ impl<'a> Estimator<'a> { ) => match est.cols.get(&c.id) { Some(col) => { null_rows += col.nulls * per; + known &= col.ndv_known; match col_id { None => col_id = Some(c.id), Some(id) if id == c.id => {} @@ -338,6 +347,7 @@ impl<'a> Estimator<'a> { out.id, Col { ndv: ndv.clamp(1.0, rows), + ndv_known: known && !other, nulls, range, }, @@ -389,6 +399,7 @@ impl<'a> Estimator<'a> { let s = stats.columns.get(&c.name).cloned().unwrap_or_default(); let col = Col { ndv: s.distinct.map_or(rows, to_f64).clamp(1.0, rows), + ndv_known: s.distinct.is_some(), nulls: s.nulls.map_or(0.0, |n| (to_f64(n) / rows).min(1.0)), range: s .min @@ -431,6 +442,10 @@ fn column_ndv(e: &Expr, est: &Est) -> Option { column(e, est).map(|c| c.ndv) } +fn is_null_literal(e: &Expr) -> bool { + matches!(literal(e), Some(Literal::Null(_))) +} + fn literal(e: &Expr) -> Option<&Literal> { match &e.node { ExprNode::Literal(l) => Some(l), @@ -462,10 +477,27 @@ pub(crate) fn selectivity(pred: &Expr, est: &Est) -> f64 { x + y - x * y } (Some((Builtin::Not, [a], _)), _) => 1.0 - selectivity(a, est), + // `c <=> NULL` holds exactly where `c` is NULL, as `c IS NULL`. + (Some((Builtin::EqNullSafe, [a, b], _)), _) if is_null_literal(a) || is_null_literal(b) => { + let c = if is_null_literal(a) { b } else { a }; + column(c, est).map_or(DEFAULT_SELECTIVITY, |c| c.nulls) + } (Some((Builtin::Eq | Builtin::EqNullSafe, [a, b], _)), _) => { match (column(a, est), column(b, est)) { (Some(x), Some(y)) => 1.0 / x.ndv.max(y.ndv), - (Some(x), None) | (None, Some(x)) => 1.0 / x.ndv, + // `col = literal`: one value of the column's. Without a + // distinct count, `1 / rows` would claim every value is + // unique (one matching row); a low-cardinality column + // (`l_returnflag = 'R'`) would then look tiny and be + // built. + (Some(x), None) | (None, Some(x)) => { + let with_literal = literal(a).is_some() || literal(b).is_some(); + if with_literal && !x.ndv_known { + DEFAULT_SELECTIVITY + } else { + 1.0 / x.ndv + } + } (None, None) => DEFAULT_SELECTIVITY, } } @@ -480,8 +512,8 @@ pub(crate) fn selectivity(pred: &Expr, est: &Est) -> f64 { } } (Some((Builtin::In, [v, list @ ..], _)), _) => match column(v, est) { - Some(c) => to_f64(u64::try_from(list.len()).unwrap_or(u64::MAX)) / c.ndv, - None => DEFAULT_SELECTIVITY, + Some(c) if c.ndv_known => to_f64(u64::try_from(list.len()).unwrap_or(u64::MAX)) / c.ndv, + _ => DEFAULT_SELECTIVITY, }, (Some((Builtin::IsNull, [a], _)), _) => { column(a, est).map_or(DEFAULT_SELECTIVITY, |c| c.nulls) diff --git a/crates/krishiv-optimizer/tests/it/join_order.rs b/crates/krishiv-optimizer/tests/it/join_order.rs index a2aebd2a0..a0e74b8f2 100644 --- a/crates/krishiv-optimizer/tests/it/join_order.rs +++ b/crates/krishiv-optimizer/tests/it/join_order.rs @@ -317,6 +317,56 @@ const TPCDS_SF1: &[(&str, u64, &[Column])] = &[ ("ca_country", "string", f64::NAN, f64::NAN, 1_525, None), ], ), + // Footer statistics as Parquet gives them (TPC-DS SF1): row counts and + // bounds; distinct counts only from integral bounds (`max - min + 1`, + // at most the non-null rows), none for strings. + ( + "catalog_sales", + 1_441_548, + &[ + ( + "cs_sold_date_sk", + "bigint", + 2_450_815.0, + 2_452_648.0, + 7_191, + Some(1_834), + ), + ( + "cs_bill_cdemo_sk", + "bigint", + 5.0, + 1_920_765.0, + 7_183, + Some(1_434_365), + ), + ], + ), + ( + "date_dim", + 73_049, + &[ + ( + "d_date_sk", + "bigint", + 2_415_022.0, + 2_488_070.0, + 0, + Some(73_049), + ), + ("d_year", "bigint", 1_900.0, 2_100.0, 0, Some(201)), + ], + ), + ( + "customer_demographics", + 1_920_800, + &[ + ("cd_demo_sk", "bigint", 1.0, 1_920_800.0, 0, Some(1_920_800)), + ("cd_gender", "string", f64::NAN, f64::NAN, 0, None), + ("cd_marital_status", "string", f64::NAN, f64::NAN, 0, None), + ("cd_education_status", "string", f64::NAN, f64::NAN, 0, None), + ], + ), ]; #[derive(Debug)] @@ -424,3 +474,176 @@ fn implied_filters_of_on_conditions_keep_the_rows() { "SELECT t.a, u.d FROM t LEFT JOIN u ON (t.b = 'x' AND u.d = 'y') OR (t.b = 'z' AND u.a > 2)", ); } + +/// The fixture's statistics without distinct counts for the columns +/// `keep` does not name, as Parquet footers give them (footers carry no +/// distinct counts). +fn without_distinct(keep: &[&str]) -> Fixture { + let mut f = fixture(); + for t in f.0.values_mut() { + for (name, c) in &mut t.columns { + if !keep.contains(&name.as_str()) { + c.distinct = None; + } + } + } + f +} + +/// `col = literal` and `col IN (...)` read the column's distinct count +/// when one is known; without one the selectivity is the default, not +/// `1 / rows` (which would estimate a single matching row for +/// `l_returnflag = 'R'` and make 60M rows look like the smallest input). +#[test] +fn literal_equalities_without_a_distinct_count_use_the_default() { + let ctx = support::ctx(); + let rows = |sql: &str, stats: &Fixture| { + krishiv_optimizer::estimated_rows(support::resolve(sql, &ctx).root(), stats) + }; + let known = fixture(); + let footers = without_distinct(&[]); + let lineitem = 59_986_052.0; + let close = |a: f64, b: f64| (a - b).abs() <= 1e-6 * b; + let eq = "SELECT l_orderkey FROM lineitem WHERE l_returnflag = 'R'"; + assert!( + close(rows(eq, &known), lineitem / 3.0), + "{}", + rows(eq, &known) + ); + assert!( + close(rows(eq, &footers), lineitem * 0.2), + "{}", + rows(eq, &footers) + ); + // The literal on the left reads the same. + let flipped = "SELECT l_orderkey FROM lineitem WHERE 'R' = l_returnflag"; + assert!(close(rows(flipped, &footers), lineitem * 0.2)); + let within = "SELECT l_orderkey FROM lineitem WHERE l_shipmode IN ('MAIL', 'SHIP')"; + assert!(close(rows(within, &known), lineitem * 2.0 / 7.0)); + assert!(close(rows(within, &footers), lineitem * 0.2)); + // A distinct count from statistics still counts without the others. + let only = without_distinct(&["l_returnflag"]); + assert!(close(rows(eq, &only), lineitem / 3.0)); + // An equality between two columns of one input keeps `1 / ndv`. + // `c <=> NULL` is `c IS NULL`: the column's null fraction (none here). + // Optimized first: folding makes the coerced NULL a literal. + let folded = |sql: &str, stats: &Fixture| { + let plan = optimize_with(&support::resolve(sql, &ctx), stats).unwrap(); + krishiv_optimizer::estimated_rows(plan.root(), stats) + }; + let is_null = folded( + "SELECT l_orderkey FROM lineitem WHERE l_returnflag IS NULL", + &known, + ); + for sql in [ + "SELECT l_orderkey FROM lineitem WHERE l_returnflag <=> NULL", + "SELECT l_orderkey FROM lineitem WHERE NULL <=> l_returnflag", + ] { + let (k, f) = (folded(sql, &known), folded(sql, &footers)); + assert!( + close(k, is_null) && close(f, is_null), + "{sql}: {k} {f} vs {is_null}" + ); + } + let cols = "SELECT l_orderkey FROM lineitem WHERE l_commitdate = l_receiptdate"; + assert!(close(rows(cols, &footers), 1.0), "{}", rows(cols, &footers)); +} + +/// TPC-DS q26's shape: a fact table joined to a dimension filtered to one +/// year and to one filtered by three string equalities. A join with a +/// filtered input counts the distinct keys the filter keeps (at most its +/// rows), so the year of dates joins about a fifth of the fact rows (the +/// fact holds five years), not 1/200 of them as the whole 73,049-date +/// domain would say. The demographics join, the more selective one, then +/// comes first and the fact table stays the larger (probe) input of every +/// join, so each dimension's build can filter the fact scan. +#[test] +fn filtered_dimensions_count_the_keys_they_keep() { + let sql = "SELECT count(*) FROM catalog_sales, customer_demographics, date_dim \ + WHERE cs_sold_date_sk = d_date_sk AND cs_bill_cdemo_sk = cd_demo_sk \ + AND cd_gender = 'M' AND cd_marital_status = 'S' \ + AND cd_education_status = 'College' AND d_year = 2000"; + let plan = support::resolve(sql, &tpcds_ctx()); + let optimized = optimize_with(&plan, &TpcdsFooters).unwrap(); + let text = optimized.root().to_string(); + let line = |needle: &str| { + text.lines() + .position(|l| l.contains("Join Inner") && l.contains(needle)) + .unwrap_or_else(|| panic!("no join on {needle}:\n{text}")) + }; + // The date join is above (printed before) the demographics join. + assert!(line("d_date_sk") < line("cd_demo_sk"), "{text}"); + insta::assert_snapshot!(text); +} + +/// TPC-DS q13's shape: one `or` over three inputs whose disjuncts all hold +/// the same key equalities. The equalities are join conditions of their +/// own, so the inputs meet in equi-joins; as one condition over all three +/// inputs they could only meet in a cross product (q13 crossed a filtered +/// customer_demographics with the fact side). +#[test] +fn equalities_every_disjunct_holds_join_their_inputs() { + let sql = "SELECT count(*) FROM lineitem, orders, customer \ + WHERE (l_orderkey = o_orderkey AND o_custkey = c_custkey \ + AND l_quantity > 10 AND c_acctbal > 0) \ + OR (l_orderkey = o_orderkey AND o_custkey = c_custkey \ + AND l_quantity < 5 AND o_totalprice > 100)"; + let plan = support::resolve(sql, &support::ctx()); + let optimized = optimize_with(&plan, &fixture()).unwrap(); + no_cross_product(optimized.root(), sql); + insta::assert_snapshot!(optimized.root().to_string()); + // The same rows, and a disjunct of only shared equalities leaves just + // the equalities. + same_rows( + "SELECT t.a, u.d, x.b FROM t, u, t x \ + WHERE (t.a = u.a AND u.a = x.a AND t.b = 'p') OR (t.a = u.a AND u.a = x.a AND x.c > 1)", + ); + let only = same_rows( + "SELECT t.a, u.d, x.b FROM t, u, t x \ + WHERE (t.a = u.a AND u.a = x.a) OR (t.a = u.a AND u.a = x.a AND x.c > 1)", + ); + assert!(!only.root().to_string().contains(" or "), "{}", only.root()); + // Equalities only some disjuncts hold stay inside the `or`. + same_rows("SELECT t.a, u.d FROM t, u WHERE (t.a = u.a AND t.b = 'p') OR (t.c = u.a)"); +} + +/// TPC-DS q85's shape: an `or` over three inputs that other conditions +/// already join. Its shared equality is not needed to connect them, so it +/// stays inside the `or` instead of becoming an extra join key (q85 would +/// otherwise join its two demographics instances on two string columns). +#[test] +fn shared_equalities_stay_in_the_or_when_the_inputs_are_joined() { + let sql = "SELECT count(*) FROM lineitem, orders, customer \ + WHERE l_orderkey = o_orderkey AND o_custkey = c_custkey \ + AND ((o_orderpriority = c_mktsegment AND l_quantity > 10) \ + OR (o_orderpriority = c_mktsegment AND l_quantity < 5))"; + let plan = support::resolve(sql, &support::ctx()); + let optimized = optimize_with(&plan, &fixture()).unwrap(); + let text = optimized.root().to_string(); + assert_eq!(text.matches("(o_orderpriority#").count(), 2, "{text}"); + insta::assert_snapshot!(text); +} + +/// A grouping-set column passed through `Expand` keeps whether its +/// distinct count comes from statistics, so `col = literal` above a +/// `ROLLUP` reads the count when the statistics have one and the default +/// otherwise. +#[test] +fn expanded_columns_keep_whether_their_distinct_count_is_known() { + let ctx = support::ctx(); + let rows = |sql: &str, stats: &Fixture| { + krishiv_optimizer::estimated_rows(support::resolve(sql, &ctx).root(), stats) + }; + let all = "SELECT * FROM (SELECT l_returnflag, l_orderkey FROM lineitem \ + GROUP BY ROLLUP (l_returnflag, l_orderkey))"; + let eq = "SELECT * FROM (SELECT l_returnflag, l_orderkey FROM lineitem \ + GROUP BY ROLLUP (l_returnflag, l_orderkey)) WHERE l_returnflag = 'R'"; + let (known, footers) = (fixture(), without_distinct(&[])); + let close = |a: f64, b: f64| (a - b).abs() <= 1e-6 * b; + // Three return flags in the statistics: a third of the rows. + let k = rows(eq, &known) / rows(all, &known); + assert!(close(k, 1.0 / 3.0), "{k}"); + // No distinct count: the default selectivity. + let f = rows(eq, &footers) / rows(all, &footers); + assert!(close(f, krishiv_optimizer::DEFAULT_SELECTIVITY), "{f}"); +} diff --git a/crates/krishiv-optimizer/tests/it/props.rs b/crates/krishiv-optimizer/tests/it/props.rs index 4e431c2e4..313771348 100644 --- a/crates/krishiv-optimizer/tests/it/props.rs +++ b/crates/krishiv-optimizer/tests/it/props.rs @@ -7,7 +7,7 @@ use std::collections::HashSet; use std::sync::Arc; -use krishiv_optimizer::optimize; +use krishiv_optimizer::{optimize, optimize_with}; use krishiv_plan::{ AttrId, Builtin, Expr, ExprNode, FunctionId, Grouping, JoinCondition, JoinKind, LogicalPlan, PlanNode, ResolvedExpr, @@ -487,8 +487,8 @@ fn restriction_after_join_ordering_reaches_a_fixed_point() { ORDER BY c0, c1 LIMIT 4"; let ctx = support::ctx(); let plan = support::resolve(sql, &ctx); - let optimized = optimize(&plan).unwrap(); - let again = optimize(&optimized).unwrap(); + let optimized = optimize_with(&plan, &support::AllDistinct).unwrap(); + let again = optimize_with(&optimized, &support::AllDistinct).unwrap(); assert!( again.structurally_eq(&optimized), "not a fixed point:\n{}\n{}", diff --git a/crates/krishiv-optimizer/tests/it/rules.rs b/crates/krishiv-optimizer/tests/it/rules.rs index 1aeb3f0b7..260eb5e8e 100644 --- a/crates/krishiv-optimizer/tests/it/rules.rs +++ b/crates/krishiv-optimizer/tests/it/rules.rs @@ -7,7 +7,7 @@ use std::collections::HashMap; use std::sync::Arc; use krishiv_common::error::ErrorClass; -use krishiv_optimizer::{ColumnStats, Statistics, TableStats, TopN, optimize}; +use krishiv_optimizer::{ColumnStats, Statistics, TableStats, TopN, optimize, optimize_with}; use krishiv_plan::{ Builtin, Expr, ExprNode, FunctionId, FunctionKind, Literal, LogicalPlan, Plan, PlanNode, ResolvedExpr, ResolvedFunction, Scan, @@ -423,13 +423,13 @@ fn rules() { let ctx = support::ctx(); for (name, sql) in CASES { let plan = support::resolve(sql, &ctx); - let optimized = optimize(&plan).unwrap(); + let optimized = optimize_with(&plan, &support::AllDistinct).unwrap(); assert_eq!( optimized.root().resolved_output(), plan.root().resolved_output(), "{name} keeps its output" ); - let again = optimize(&optimized).unwrap(); + let again = optimize_with(&optimized, &support::AllDistinct).unwrap(); assert!( again.structurally_eq(&optimized), "{name} is a fixed point:\n{}", diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__equalities_every_disjunct_holds_join_their_inputs.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__equalities_every_disjunct_holds_join_their_inputs.snap new file mode 100644 index 000000000..4ebeb28ce --- /dev/null +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__equalities_every_disjunct_holds_join_their_inputs.snap @@ -0,0 +1,13 @@ +--- +source: crates/krishiv-optimizer/tests/it/join_order.rs +expression: optimized.root().to_string() +--- +Aggregate [], [count(1) AS count(1)#33] ++- Project [l_orderkey#0] + +- Join Inner, ((l_orderkey#0 == o_orderkey#16) and (((l_quantity#4 > 10.00) and (c_acctbal#30 > 0.00)) or ((l_quantity#4 < 5.00) and (o_totalprice#19 > 100.00)))) + :- Filter ((l_quantity#4 > 10.00) or (l_quantity#4 < 5.00)) + : +- Scan spark_catalog.default.lineitem [l_orderkey#0, l_quantity#4] + +- Project [o_orderkey#16, o_totalprice#19, c_acctbal#30] + +- Join Inner, (o_custkey#17 == c_custkey#25) + :- Scan spark_catalog.default.orders [o_orderkey#16, o_custkey#17, o_totalprice#19] + +- Scan spark_catalog.default.customer [c_custkey#25, c_acctbal#30] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__filtered_dimensions_count_the_keys_they_keep.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__filtered_dimensions_count_the_keys_they_keep.snap new file mode 100644 index 000000000..6c6616f51 --- /dev/null +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__filtered_dimensions_count_the_keys_they_keep.snap @@ -0,0 +1,16 @@ +--- +source: crates/krishiv-optimizer/tests/it/join_order.rs +expression: text +--- +Aggregate [], [count(1) AS count(1)#8] ++- Project [cs_sold_date_sk#0] + +- Join Inner, (cs_sold_date_sk#0 == d_date_sk#6) + :- Project [cs_sold_date_sk#0] + : +- Join Inner, (cs_bill_cdemo_sk#1 == cd_demo_sk#2) + : :- Scan spark_catalog.default.catalog_sales [cs_sold_date_sk#0, cs_bill_cdemo_sk#1] + : +- Project [cd_demo_sk#2] + : +- Filter (((cd_gender#3 == 'M') and (cd_marital_status#4 == 'S')) and (cd_education_status#5 == 'College')) + : +- Scan spark_catalog.default.customer_demographics [cd_demo_sk#2, cd_gender#3, cd_marital_status#4, cd_education_status#5] + +- Project [d_date_sk#6] + +- Filter (d_year#7 == 2000L) + +- Scan spark_catalog.default.date_dim [d_date_sk#6, d_year#7] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__large_trees_are_ordered_greedily.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__large_trees_are_ordered_greedily.snap index 236e82895..51f6c65f8 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__large_trees_are_ordered_greedily.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__large_trees_are_ordered_greedily.snap @@ -3,60 +3,60 @@ source: crates/krishiv-optimizer/tests/it/join_order.rs expression: optimized.root().to_string() --- Project [b#34] -+- Join Inner, (CAST(a#33 AS BIGINT) == c#32) - :- Project [a#33, b#34] - : +- Shift +33 - : +- Scan spark_catalog.default.t [a#0, b#1, c#2] (shared #1) - +- Project [c#32] - +- Join Inner, (CAST(a#30 AS BIGINT) == c#29) - :- Project [a#30, c#32] - : +- Shift +30 - : +- Shared #1, printed above - +- Project [c#29] - +- Join Inner, (CAST(a#27 AS BIGINT) == c#26) - :- Project [a#27, c#29] - : +- Shift +27 - : +- Shared #1, printed above - +- Project [c#26] - +- Join Inner, (CAST(a#24 AS BIGINT) == c#23) - :- Project [a#24, c#26] - : +- Shift +24 - : +- Shared #1, printed above - +- Project [c#23] - +- Join Inner, (CAST(a#21 AS BIGINT) == c#20) - :- Project [a#21, c#23] - : +- Shift +21 - : +- Shared #1, printed above - +- Project [c#20] - +- Join Inner, (CAST(a#18 AS BIGINT) == c#17) - :- Project [a#18, c#20] - : +- Shift +18 - : +- Shared #1, printed above - +- Project [c#17] - +- Join Inner, (CAST(a#15 AS BIGINT) == c#14) - :- Project [a#15, c#17] - : +- Shift +15 - : +- Shared #1, printed above - +- Project [c#14] - +- Join Inner, (CAST(a#12 AS BIGINT) == c#11) - :- Project [a#12, c#14] - : +- Shift +12 - : +- Shared #1, printed above - +- Project [c#11] - +- Join Inner, (CAST(a#9 AS BIGINT) == c#8) - :- Project [a#9, c#11] - : +- Shift +9 - : +- Shared #1, printed above - +- Project [c#8] - +- Join Inner, (CAST(a#6 AS BIGINT) == c#5) - :- Project [a#6, c#8] - : +- Shift +6 - : +- Shared #1, printed above - +- Project [c#5] - +- Join Inner, (CAST(a#3 AS BIGINT) == c#2) - :- Project [a#3, c#5] - : +- Shift +3 - : +- Shared #1, printed above - +- Project [c#2] - +- Filter (b#1 == 'x') - +- Shared #1, printed above ++- Join Inner, (CAST(a#3 AS BIGINT) == c#2) + :- Project [a#3, b#34] + : +- Join Inner, (CAST(a#33 AS BIGINT) == c#32) + : :- Project [a#3, c#32] + : : +- Join Inner, (CAST(a#30 AS BIGINT) == c#29) + : : :- Project [a#3, c#29] + : : : +- Join Inner, (CAST(a#27 AS BIGINT) == c#26) + : : : :- Project [a#3, c#26] + : : : : +- Join Inner, (CAST(a#24 AS BIGINT) == c#23) + : : : : :- Project [a#3, c#23] + : : : : : +- Join Inner, (CAST(a#21 AS BIGINT) == c#20) + : : : : : :- Project [a#3, c#20] + : : : : : : +- Join Inner, (CAST(a#18 AS BIGINT) == c#17) + : : : : : : :- Project [a#3, c#17] + : : : : : : : +- Join Inner, (CAST(a#15 AS BIGINT) == c#14) + : : : : : : : :- Project [a#3, c#14] + : : : : : : : : +- Join Inner, (CAST(a#12 AS BIGINT) == c#11) + : : : : : : : : :- Project [a#3, c#11] + : : : : : : : : : +- Join Inner, (CAST(a#9 AS BIGINT) == c#8) + : : : : : : : : : :- Project [a#3, c#8] + : : : : : : : : : : +- Join Inner, (CAST(a#6 AS BIGINT) == c#5) + : : : : : : : : : : :- Project [a#3, c#5] + : : : : : : : : : : : +- Shift +3 + : : : : : : : : : : : +- Scan spark_catalog.default.t [a#0, b#1, c#2] (shared #1) + : : : : : : : : : : +- Project [a#6, c#8] + : : : : : : : : : : +- Shift +6 + : : : : : : : : : : +- Shared #1, printed above + : : : : : : : : : +- Project [a#9, c#11] + : : : : : : : : : +- Shift +9 + : : : : : : : : : +- Shared #1, printed above + : : : : : : : : +- Project [a#12, c#14] + : : : : : : : : +- Shift +12 + : : : : : : : : +- Shared #1, printed above + : : : : : : : +- Project [a#15, c#17] + : : : : : : : +- Shift +15 + : : : : : : : +- Shared #1, printed above + : : : : : : +- Project [a#18, c#20] + : : : : : : +- Shift +18 + : : : : : : +- Shared #1, printed above + : : : : : +- Project [a#21, c#23] + : : : : : +- Shift +21 + : : : : : +- Shared #1, printed above + : : : : +- Project [a#24, c#26] + : : : : +- Shift +24 + : : : : +- Shared #1, printed above + : : : +- Project [a#27, c#29] + : : : +- Shift +27 + : : : +- Shared #1, printed above + : : +- Project [a#30, c#32] + : : +- Shift +30 + : : +- Shared #1, printed above + : +- Project [a#33, b#34] + : +- Shift +33 + : +- Shared #1, printed above + +- Project [c#2] + +- Filter (b#1 == 'x') + +- Shared #1, printed above diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__positional_roots_keep_their_columns.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__positional_roots_keep_their_columns.snap index cd26fc387..b17196efe 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__positional_roots_keep_their_columns.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__positional_roots_keep_their_columns.snap @@ -3,11 +3,11 @@ source: crates/krishiv-optimizer/tests/it/join_order.rs expression: optimized.root().to_string() --- Project [a#0, b#1, c#2, a#3, d#4, a#5, b#6, c#7] -+- Join Inner, (c#7 == CAST(a#3 AS BIGINT)) - :- Shift +5 ++- Join Inner, (a#0 == a#3) + :- Filter (b#1 == 'y') : +- Scan spark_catalog.default.t [a#0, b#1, c#2] (shared #1) - +- Join Inner, (a#0 == a#3) - :- Filter (b#1 == 'y') + +- Join Inner, (c#7 == CAST(a#3 AS BIGINT)) + :- Shift +5 : +- Shared #1, printed above +- Filter (d#4 == 'z') +- Scan spark_catalog.default.u [a#3, d#4] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__shared_equalities_stay_in_the_or_when_the_inputs_are_joined.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__shared_equalities_stay_in_the_or_when_the_inputs_are_joined.snap new file mode 100644 index 000000000..98a4f5deb --- /dev/null +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__join_order__shared_equalities_stay_in_the_or_when_the_inputs_are_joined.snap @@ -0,0 +1,13 @@ +--- +source: crates/krishiv-optimizer/tests/it/join_order.rs +expression: text +--- +Aggregate [], [count(1) AS count(1)#33] ++- Project [l_orderkey#0] + +- Join Inner, ((((o_orderpriority#21 == c_mktsegment#31) and (l_quantity#4 > 10.00)) or ((o_orderpriority#21 == c_mktsegment#31) and (l_quantity#4 < 5.00))) and (l_orderkey#0 == o_orderkey#16)) + :- Filter ((l_quantity#4 > 10.00) or (l_quantity#4 < 5.00)) + : +- Scan spark_catalog.default.lineitem [l_orderkey#0, l_quantity#4] + +- Project [o_orderkey#16, o_orderpriority#21, c_mktsegment#31] + +- Join Inner, (o_custkey#17 == c_custkey#25) + :- Scan spark_catalog.default.orders [o_orderkey#16, o_custkey#17, o_orderpriority#21] + +- Scan spark_catalog.default.customer [c_custkey#25, c_mktsegment#31] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q02.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q02.snap index 9913a93db..49005db6b 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q02.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q02.snap @@ -6,39 +6,44 @@ Limit 100 +- Sort [s_acctbal#14 DESC NULLS LAST, n_name#22 ASC NULLS FIRST, s_name#10 ASC NULLS FIRST, p_partkey#0 ASC NULLS FIRST], true +- Project [s_acctbal#14, s_name#10, n_name#22, p_partkey#0, p_mfgr#2, s_address#11, s_phone#13, s_comment#15] +- Join Inner, (n_regionkey#23 == r_regionkey#25) - :- Project [n_name#22, n_regionkey#23, s_name#10, s_address#11, s_phone#13, s_acctbal#14, s_comment#15, p_partkey#0, p_mfgr#2] - : +- Join Inner, (s_nationkey#12 == n_nationkey#21) - : :- Scan spark_catalog.default.nation [n_nationkey#21, n_name#22, n_regionkey#23] (shared #1) - : +- Project [s_name#10, s_address#11, s_nationkey#12, s_phone#13, s_acctbal#14, s_comment#15, p_partkey#0, p_mfgr#2] - : +- Join Inner, (s_suppkey#9 == ps_suppkey#17) - : :- Scan spark_catalog.default.supplier [s_suppkey#9, s_name#10, s_address#11, s_nationkey#12, s_phone#13, s_acctbal#14, s_comment#15] (shared #2) - : +- Project [ps_suppkey#17, p_partkey#0, p_mfgr#2] - : +- Join Inner, ((ps_supplycost#19 == min(ps_supplycost)#47) and (p_partkey#0 == ps_partkey#16)) - : :- Scan spark_catalog.default.partsupp [ps_partkey#16, ps_suppkey#17, ps_supplycost#19] (shared #3) - : +- Project [p_partkey#0, p_mfgr#2, min(ps_supplycost)#47] - : +- Join Inner, (p_partkey#0 == ps_partkey#28) - : :- Project [p_partkey#0, p_mfgr#2] - : : +- Filter ((p_size#5 == 15) and (p_type#4 like '%BRASS')) - : : +- Scan spark_catalog.default.part [p_partkey#0, p_mfgr#2, p_type#4, p_size#5] - : +- Aggregate [ps_partkey#28], [min(ps_supplycost#31) AS min(ps_supplycost)#47, ps_partkey#28] - : +- Project [ps_partkey#28, ps_supplycost#31] - : +- Join Inner, (s_suppkey#33 == ps_suppkey#29) - : :- Shift +12 - : : +- Shared #3, printed above - : +- Project [s_suppkey#33] - : +- Join Inner, (s_nationkey#36 == n_nationkey#40) - : :- Project [s_suppkey#33, s_nationkey#36] - : : +- Shift +24 - : : +- Shared #2, printed above - : +- Project [n_nationkey#40] - : +- Join Inner, (n_regionkey#42 == r_regionkey#44) - : :- Project [n_nationkey#40, n_regionkey#42] - : : +- Shift +19 - : : +- Shared #1, printed above - : +- Project [r_regionkey#44] - : +- Shift +19 - : +- Filter (r_name#26 == 'EUROPE') - : +- Scan spark_catalog.default.region [r_regionkey#25, r_name#26] (shared #4) - +- Project [r_regionkey#25] - +- Filter (r_name#26 == 'EUROPE') - +- Shared #4, printed above + :- Project [r_regionkey#25] + : +- Filter (r_name#26 == 'EUROPE') + : +- Scan spark_catalog.default.region [r_regionkey#25, r_name#26] (shared #1) + +- Project [n_name#22, n_regionkey#23, s_name#10, s_address#11, s_phone#13, s_acctbal#14, s_comment#15, p_partkey#0, p_mfgr#2] + +- Join Inner, (s_nationkey#12 == n_nationkey#21) + :- Scan spark_catalog.default.nation [n_nationkey#21, n_name#22, n_regionkey#23] (shared #2) + +- Project [s_name#10, s_address#11, s_nationkey#12, s_phone#13, s_acctbal#14, s_comment#15, p_partkey#0, p_mfgr#2] + +- Join Inner, (s_suppkey#9 == ps_suppkey#17) + :- Scan spark_catalog.default.supplier [s_suppkey#9, s_name#10, s_address#11, s_nationkey#12, s_phone#13, s_acctbal#14, s_comment#15] (shared #3) + +- Project [ps_suppkey#17, p_partkey#0, p_mfgr#2] + +- Join Inner, ((ps_supplycost#19 == min(ps_supplycost)#47) and (p_partkey#0 == ps_partkey#16)) + :- Scan spark_catalog.default.partsupp [ps_partkey#16, ps_suppkey#17, ps_supplycost#19] (shared #4) + +- Project [min(ps_supplycost)#47, p_partkey#0, p_mfgr#2] + +- Join Inner, (p_partkey#0 == ps_partkey#28) + :- Aggregate [ps_partkey#28], [min(ps_supplycost#31) AS min(ps_supplycost)#47, ps_partkey#28] + : +- Join LeftSemi, (ps_partkey#28 == p_partkey#48) + : :- Project [ps_partkey#28, ps_supplycost#31] + : : +- Join Inner, (s_suppkey#33 == ps_suppkey#29) + : : :- Shift +12 + : : : +- Shared #4, printed above + : : +- Project [s_suppkey#33] + : : +- Join Inner, (s_nationkey#36 == n_nationkey#40) + : : :- Project [s_suppkey#33, s_nationkey#36] + : : : +- Shift +24 + : : : +- Shared #3, printed above + : : +- Project [n_nationkey#40] + : : +- Join Inner, (n_regionkey#42 == r_regionkey#44) + : : :- Project [n_nationkey#40, n_regionkey#42] + : : : +- Shift +19 + : : : +- Shared #2, printed above + : : +- Project [r_regionkey#44] + : : +- Shift +19 + : : +- Filter (r_name#26 == 'EUROPE') + : : +- Shared #1, printed above + : +- Shift +48 + : +- Project [p_partkey#0] + : +- Filter ((p_size#5 == 15) and (p_type#4 like '%BRASS')) + : +- Scan spark_catalog.default.part [p_partkey#0, p_mfgr#2, p_type#4, p_size#5] (shared #5) + +- Project [p_partkey#0, p_mfgr#2] + +- Filter ((p_size#5 == 15) and (p_type#4 like '%BRASS')) + +- Shared #5, printed above diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q05.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q05.snap index f3fc44e1e..66f133a33 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q05.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q05.snap @@ -4,23 +4,23 @@ expression: optimized.root().to_string() --- Sort [revenue#47 DESC NULLS LAST], true +- Aggregate [n_name#41], [n_name#41, sum((l_extendedprice#22 * (1 - l_discount#23))) AS revenue#47] - +- Project [l_extendedprice#22, l_discount#23, n_name#41] - +- Join Inner, ((l_orderkey#17 == o_orderkey#8) and (c_custkey#0 == o_custkey#9)) - :- Project [o_orderkey#8, o_custkey#9] - : +- Filter ((o_orderdate#12 >= DATE '1994-01-01') and (o_orderdate#12 < DATE '1995-01-01')) - : +- Scan spark_catalog.default.orders [o_orderkey#8, o_custkey#9, o_orderdate#12] - +- Project [l_orderkey#17, l_extendedprice#22, l_discount#23, c_custkey#0, n_name#41] - +- Join Inner, (l_suppkey#19 == s_suppkey#33) - :- Scan spark_catalog.default.lineitem [l_orderkey#17, l_suppkey#19, l_extendedprice#22, l_discount#23] - +- Project [c_custkey#0, s_suppkey#33, n_name#41] - +- Join Inner, (c_nationkey#3 == s_nationkey#36) - :- Scan spark_catalog.default.customer [c_custkey#0, c_nationkey#3] - +- Project [s_suppkey#33, s_nationkey#36, n_name#41] - +- Join Inner, (s_nationkey#36 == n_nationkey#40) + +- Project [n_name#41, l_extendedprice#22, l_discount#23] + +- Join Inner, (n_regionkey#42 == r_regionkey#44) + :- Project [r_regionkey#44] + : +- Filter (r_name#45 == 'ASIA') + : +- Scan spark_catalog.default.region [r_regionkey#44, r_name#45] + +- Project [n_name#41, n_regionkey#42, l_extendedprice#22, l_discount#23] + +- Join Inner, (s_nationkey#36 == n_nationkey#40) + :- Scan spark_catalog.default.nation [n_nationkey#40, n_name#41, n_regionkey#42] + +- Project [l_extendedprice#22, l_discount#23, s_nationkey#36] + +- Join Inner, ((l_suppkey#19 == s_suppkey#33) and (l_orderkey#17 == o_orderkey#8)) + :- Scan spark_catalog.default.lineitem [l_orderkey#17, l_suppkey#19, l_extendedprice#22, l_discount#23] + +- Project [s_suppkey#33, s_nationkey#36, o_orderkey#8] + +- Join Inner, (c_nationkey#3 == s_nationkey#36) :- Scan spark_catalog.default.supplier [s_suppkey#33, s_nationkey#36] - +- Project [n_nationkey#40, n_name#41] - +- Join Inner, (n_regionkey#42 == r_regionkey#44) - :- Scan spark_catalog.default.nation [n_nationkey#40, n_name#41, n_regionkey#42] - +- Project [r_regionkey#44] - +- Filter (r_name#45 == 'ASIA') - +- Scan spark_catalog.default.region [r_regionkey#44, r_name#45] + +- Project [c_nationkey#3, o_orderkey#8] + +- Join Inner, (c_custkey#0 == o_custkey#9) + :- Scan spark_catalog.default.customer [c_custkey#0, c_nationkey#3] + +- Project [o_orderkey#8, o_custkey#9] + +- Filter ((o_orderdate#12 >= DATE '1994-01-01') and (o_orderdate#12 < DATE '1995-01-01')) + +- Scan spark_catalog.default.orders [o_orderkey#8, o_custkey#9, o_orderdate#12] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q07.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q07.snap index 0297bd5ec..67bd79f94 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q07.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q07.snap @@ -5,21 +5,22 @@ expression: optimized.root().to_string() Sort [supp_nation#48 ASC NULLS FIRST, cust_nation#49 ASC NULLS FIRST, l_year#50 ASC NULLS FIRST], true +- Aggregate [supp_nation#48, cust_nation#49, l_year#50], [supp_nation#48, cust_nation#49, l_year#50, sum(volume#51) AS revenue#52] +- Project [n_name#41 AS supp_nation#48, n_name#45 AS cust_nation#49, extract('year', l_shipdate#17) AS l_year#50, (l_extendedprice#12 * (1 - l_discount#13)) AS volume#51] - +- Join Inner, ((c_custkey#32 == o_custkey#24) and (o_orderkey#23 == l_orderkey#7)) - :- Scan spark_catalog.default.orders [o_orderkey#23, o_custkey#24] - +- Project [c_custkey#32, n_name#45, l_orderkey#7, l_extendedprice#12, l_discount#13, l_shipdate#17, n_name#41] - +- Join Inner, (c_nationkey#35 == n_nationkey#44) + +- Join Inner, ((c_nationkey#35 == n_nationkey#44) and (((n_name#41 == 'FRANCE') and (n_name#45 == 'GERMANY')) or ((n_name#41 == 'GERMANY') and (n_name#45 == 'FRANCE')))) + :- Shift +4 + : +- Filter ((n_name#41 == 'GERMANY') or (n_name#41 == 'FRANCE')) + : +- Scan spark_catalog.default.nation [n_nationkey#40, n_name#41] (shared #1) + +- Project [c_nationkey#35, n_name#41, l_extendedprice#12, l_discount#13, l_shipdate#17] + +- Join Inner, (c_custkey#32 == o_custkey#24) :- Scan spark_catalog.default.customer [c_custkey#32, c_nationkey#35] - +- Join Inner, (((n_name#41 == 'FRANCE') and (n_name#45 == 'GERMANY')) or ((n_name#41 == 'GERMANY') and (n_name#45 == 'FRANCE'))) - :- Shift +4 - : +- Filter ((n_name#41 == 'GERMANY') or (n_name#41 == 'FRANCE')) - : +- Scan spark_catalog.default.nation [n_nationkey#40, n_name#41] (shared #1) - +- Project [l_orderkey#7, l_extendedprice#12, l_discount#13, l_shipdate#17, n_name#41] - +- Join Inner, (s_suppkey#0 == l_suppkey#9) - :- Filter ((l_shipdate#17 >= DATE '1995-01-01') and (l_shipdate#17 <= DATE '1996-12-31')) - : +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_extendedprice#12, l_discount#13, l_shipdate#17] - +- Project [s_suppkey#0, n_name#41] - +- Join Inner, (s_nationkey#3 == n_nationkey#40) - :- Scan spark_catalog.default.supplier [s_suppkey#0, s_nationkey#3] - +- Filter ((n_name#41 == 'FRANCE') or (n_name#41 == 'GERMANY')) - +- Shared #1, printed above + +- Project [o_custkey#24, n_name#41, l_extendedprice#12, l_discount#13, l_shipdate#17] + +- Join Inner, (o_orderkey#23 == l_orderkey#7) + :- Scan spark_catalog.default.orders [o_orderkey#23, o_custkey#24] + +- Project [n_name#41, l_orderkey#7, l_extendedprice#12, l_discount#13, l_shipdate#17] + +- Join Inner, (s_nationkey#3 == n_nationkey#40) + :- Filter ((n_name#41 == 'FRANCE') or (n_name#41 == 'GERMANY')) + : +- Shared #1, printed above + +- Project [s_nationkey#3, l_orderkey#7, l_extendedprice#12, l_discount#13, l_shipdate#17] + +- Join Inner, (s_suppkey#0 == l_suppkey#9) + :- Scan spark_catalog.default.supplier [s_suppkey#0, s_nationkey#3] + +- Filter ((l_shipdate#17 >= DATE '1995-01-01') and (l_shipdate#17 <= DATE '1996-12-31')) + +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_extendedprice#12, l_discount#13, l_shipdate#17] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q08.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q08.snap index cbad225ca..768420300 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q08.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q08.snap @@ -9,28 +9,28 @@ Sort [o_year#60 ASC NULLS FIRST], true :- Project [n_nationkey#53, n_name#54] : +- Shift +4 : +- Scan spark_catalog.default.nation [n_nationkey#49, n_name#50, n_regionkey#51] (shared #1) - +- Project [s_nationkey#12, o_orderdate#36, l_extendedprice#21, l_discount#22] + +- Project [s_nationkey#12, l_extendedprice#21, l_discount#22, o_orderdate#36] +- Join Inner, (s_suppkey#9 == l_suppkey#18) :- Scan spark_catalog.default.supplier [s_suppkey#9, s_nationkey#12] - +- Project [o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] + +- Project [l_suppkey#18, l_extendedprice#21, l_discount#22, o_orderdate#36] +- Join Inner, (n_regionkey#51 == r_regionkey#57) - :- Project [n_regionkey#51, o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- Join Inner, (c_nationkey#44 == n_nationkey#49) - : :- Project [n_nationkey#49, n_regionkey#51] - : : +- Shared #1, printed above - : +- Project [c_nationkey#44, o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- Join Inner, (o_custkey#33 == c_custkey#41) - : :- Scan spark_catalog.default.customer [c_custkey#41, c_nationkey#44] - : +- Project [o_custkey#33, o_orderdate#36, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- Join Inner, (l_orderkey#16 == o_orderkey#32) - : :- Filter ((o_orderdate#36 >= DATE '1995-01-01') and (o_orderdate#36 <= DATE '1996-12-31')) - : : +- Scan spark_catalog.default.orders [o_orderkey#32, o_custkey#33, o_orderdate#36] - : +- Project [l_orderkey#16, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- Join Inner, (p_partkey#0 == l_partkey#17) - : :- Scan spark_catalog.default.lineitem [l_orderkey#16, l_partkey#17, l_suppkey#18, l_extendedprice#21, l_discount#22] - : +- Project [p_partkey#0] - : +- Filter (p_type#4 == 'ECONOMY ANODIZED STEEL') - : +- Scan spark_catalog.default.part [p_partkey#0, p_type#4] - +- Project [r_regionkey#57] - +- Filter (r_name#58 == 'AMERICA') - +- Scan spark_catalog.default.region [r_regionkey#57, r_name#58] + :- Project [r_regionkey#57] + : +- Filter (r_name#58 == 'AMERICA') + : +- Scan spark_catalog.default.region [r_regionkey#57, r_name#58] + +- Project [n_regionkey#51, l_suppkey#18, l_extendedprice#21, l_discount#22, o_orderdate#36] + +- Join Inner, (c_nationkey#44 == n_nationkey#49) + :- Project [n_nationkey#49, n_regionkey#51] + : +- Shared #1, printed above + +- Project [c_nationkey#44, l_suppkey#18, l_extendedprice#21, l_discount#22, o_orderdate#36] + +- Join Inner, (o_custkey#33 == c_custkey#41) + :- Scan spark_catalog.default.customer [c_custkey#41, c_nationkey#44] + +- Project [l_suppkey#18, l_extendedprice#21, l_discount#22, o_custkey#33, o_orderdate#36] + +- Join Inner, (p_partkey#0 == l_partkey#17) + :- Project [p_partkey#0] + : +- Filter (p_type#4 == 'ECONOMY ANODIZED STEEL') + : +- Scan spark_catalog.default.part [p_partkey#0, p_type#4] + +- Project [l_partkey#17, l_suppkey#18, l_extendedprice#21, l_discount#22, o_custkey#33, o_orderdate#36] + +- Join Inner, (l_orderkey#16 == o_orderkey#32) + :- Scan spark_catalog.default.lineitem [l_orderkey#16, l_partkey#17, l_suppkey#18, l_extendedprice#21, l_discount#22] + +- Filter ((o_orderdate#36 >= DATE '1995-01-01') and (o_orderdate#36 <= DATE '1996-12-31')) + +- Scan spark_catalog.default.orders [o_orderkey#32, o_custkey#33, o_orderdate#36] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q10.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q10.snap index fb9a56719..1f3afb8b2 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q10.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q10.snap @@ -5,17 +5,17 @@ expression: optimized.root().to_string() Limit 20 +- Sort [revenue#37 DESC NULLS LAST], true +- Aggregate [c_custkey#0, c_name#1, c_acctbal#5, c_phone#4, n_name#34, c_address#2, c_comment#7], [c_custkey#0, c_name#1, sum((l_extendedprice#22 * (1 - l_discount#23))) AS revenue#37, c_acctbal#5, n_name#34, c_address#2, c_phone#4, c_comment#7] - +- Project [n_name#34, c_custkey#0, c_name#1, c_address#2, c_phone#4, c_acctbal#5, c_comment#7, l_extendedprice#22, l_discount#23] - +- Join Inner, (c_nationkey#3 == n_nationkey#33) - :- Scan spark_catalog.default.nation [n_nationkey#33, n_name#34] - +- Project [c_custkey#0, c_name#1, c_address#2, c_nationkey#3, c_phone#4, c_acctbal#5, c_comment#7, l_extendedprice#22, l_discount#23] - +- Join Inner, (c_custkey#0 == o_custkey#9) - :- Scan spark_catalog.default.customer [c_custkey#0, c_name#1, c_address#2, c_nationkey#3, c_phone#4, c_acctbal#5, c_comment#7] - +- Project [o_custkey#9, l_extendedprice#22, l_discount#23] - +- Join Inner, (l_orderkey#17 == o_orderkey#8) - :- Project [o_orderkey#8, o_custkey#9] - : +- Filter ((o_orderdate#12 >= DATE '1993-10-01') and (o_orderdate#12 < DATE '1994-01-01')) - : +- Scan spark_catalog.default.orders [o_orderkey#8, o_custkey#9, o_orderdate#12] - +- Project [l_orderkey#17, l_extendedprice#22, l_discount#23] - +- Filter (l_returnflag#25 == 'R') - +- Scan spark_catalog.default.lineitem [l_orderkey#17, l_extendedprice#22, l_discount#23, l_returnflag#25] + +- Project [l_extendedprice#22, l_discount#23, n_name#34, c_custkey#0, c_name#1, c_address#2, c_phone#4, c_acctbal#5, c_comment#7] + +- Join Inner, (l_orderkey#17 == o_orderkey#8) + :- Project [l_orderkey#17, l_extendedprice#22, l_discount#23] + : +- Filter (l_returnflag#25 == 'R') + : +- Scan spark_catalog.default.lineitem [l_orderkey#17, l_extendedprice#22, l_discount#23, l_returnflag#25] + +- Project [n_name#34, c_custkey#0, c_name#1, c_address#2, c_phone#4, c_acctbal#5, c_comment#7, o_orderkey#8] + +- Join Inner, (c_nationkey#3 == n_nationkey#33) + :- Scan spark_catalog.default.nation [n_nationkey#33, n_name#34] + +- Project [c_custkey#0, c_name#1, c_address#2, c_nationkey#3, c_phone#4, c_acctbal#5, c_comment#7, o_orderkey#8] + +- Join Inner, (c_custkey#0 == o_custkey#9) + :- Scan spark_catalog.default.customer [c_custkey#0, c_name#1, c_address#2, c_nationkey#3, c_phone#4, c_acctbal#5, c_comment#7] + +- Project [o_orderkey#8, o_custkey#9] + +- Filter ((o_orderdate#12 >= DATE '1993-10-01') and (o_orderdate#12 < DATE '1994-01-01')) + +- Scan spark_catalog.default.orders [o_orderkey#8, o_custkey#9, o_orderdate#12] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q17.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q17.snap index d9c9de33d..e175ed4d7 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q17.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q17.snap @@ -5,18 +5,18 @@ expression: optimized.root().to_string() Aggregate [], [(sum(l_extendedprice#5) / 7.0) AS avg_yearly#42] +- Project [l_extendedprice#5] +- Join Inner, ((p_partkey#16 == l_partkey#26) and (CAST(l_quantity#4 AS DECIMAL(21,7)) < (0.2 * avg(l_quantity))#41)) - :- Project [l_quantity#4, l_extendedprice#5, p_partkey#16] - : +- Join Inner, (p_partkey#16 == l_partkey#1) - : :- Scan spark_catalog.default.lineitem [l_partkey#1, l_quantity#4, l_extendedprice#5] (shared #1) - : +- Project [p_partkey#16] - : +- Filter ((p_brand#19 == 'Brand#23') and (p_container#22 == 'MED BOX')) - : +- Scan spark_catalog.default.part [p_partkey#16, p_brand#19, p_container#22] (shared #2) - +- Aggregate [l_partkey#26], [(0.2 * avg(l_quantity#29)) AS (0.2 * avg(l_quantity))#41, l_partkey#26] - +- Join LeftSemi, (l_partkey#26 == p_partkey#59) - :- Project [l_partkey#26, l_quantity#29] - : +- Shift +25 - : +- Shared #1, printed above - +- Shift +43 - +- Project [p_partkey#16] - +- Filter ((p_brand#19 == 'Brand#23') and (p_container#22 == 'MED BOX')) - +- Shared #2, printed above + :- Aggregate [l_partkey#26], [(0.2 * avg(l_quantity#29)) AS (0.2 * avg(l_quantity))#41, l_partkey#26] + : +- Join LeftSemi, (l_partkey#26 == p_partkey#59) + : :- Project [l_partkey#26, l_quantity#29] + : : +- Shift +25 + : : +- Scan spark_catalog.default.lineitem [l_partkey#1, l_quantity#4, l_extendedprice#5] (shared #1) + : +- Shift +43 + : +- Project [p_partkey#16] + : +- Filter ((p_brand#19 == 'Brand#23') and (p_container#22 == 'MED BOX')) + : +- Scan spark_catalog.default.part [p_partkey#16, p_brand#19, p_container#22] (shared #2) + +- Project [l_quantity#4, l_extendedprice#5, p_partkey#16] + +- Join Inner, (p_partkey#16 == l_partkey#1) + :- Shared #1, printed above + +- Project [p_partkey#16] + +- Filter ((p_brand#19 == 'Brand#23') and (p_container#22 == 'MED BOX')) + +- Shared #2, printed above diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q19.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q19.snap index 2f3755c0c..b22844ab2 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q19.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q19.snap @@ -5,7 +5,7 @@ expression: optimized.root().to_string() Aggregate [], [sum((l_extendedprice#5 * (1 - l_discount#6))) AS revenue#25] +- Project [l_extendedprice#5, l_discount#6] +- Join Inner, ((((((p_partkey#16 == l_partkey#1) and (p_brand#19 == 'Brand#12')) and ((p_container#22 IN ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG')) and (l_quantity#4 >= 1.00))) and (((l_quantity#4 <= 11.00) and ((p_size#21 >= 1) and (p_size#21 <= 5))) and ((l_shipmode#14 IN ('AIR', 'AIR REG')) and (l_shipinstruct#13 == 'DELIVER IN PERSON')))) or ((((p_partkey#16 == l_partkey#1) and (p_brand#19 == 'Brand#23')) and ((p_container#22 IN ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK')) and (l_quantity#4 >= 10.00))) and (((l_quantity#4 <= 20.00) and ((p_size#21 >= 1) and (p_size#21 <= 10))) and ((l_shipmode#14 IN ('AIR', 'AIR REG')) and (l_shipinstruct#13 == 'DELIVER IN PERSON'))))) or ((((p_partkey#16 == l_partkey#1) and (p_brand#19 == 'Brand#34')) and ((p_container#22 IN ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG')) and (l_quantity#4 >= 20.00))) and (((l_quantity#4 <= 30.00) and ((p_size#21 >= 1) and (p_size#21 <= 15))) and ((l_shipmode#14 IN ('AIR', 'AIR REG')) and (l_shipinstruct#13 == 'DELIVER IN PERSON'))))) - :- Filter ((((((l_quantity#4 >= 1.00) and (l_quantity#4 <= 11.00)) and (l_shipmode#14 IN ('AIR', 'AIR REG'))) and (l_shipinstruct#13 == 'DELIVER IN PERSON')) or ((((l_quantity#4 >= 10.00) and (l_quantity#4 <= 20.00)) and (l_shipmode#14 IN ('AIR', 'AIR REG'))) and (l_shipinstruct#13 == 'DELIVER IN PERSON'))) or ((((l_quantity#4 >= 20.00) and (l_quantity#4 <= 30.00)) and (l_shipmode#14 IN ('AIR', 'AIR REG'))) and (l_shipinstruct#13 == 'DELIVER IN PERSON'))) - : +- Scan spark_catalog.default.lineitem [l_partkey#1, l_quantity#4, l_extendedprice#5, l_discount#6, l_shipinstruct#13, l_shipmode#14] - +- Filter ((((((p_brand#19 == 'Brand#12') and (p_container#22 IN ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG'))) and (p_size#21 >= 1)) and (p_size#21 <= 5)) or ((((p_brand#19 == 'Brand#23') and (p_container#22 IN ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK'))) and (p_size#21 >= 1)) and (p_size#21 <= 10))) or ((((p_brand#19 == 'Brand#34') and (p_container#22 IN ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG'))) and (p_size#21 >= 1)) and (p_size#21 <= 15))) - +- Scan spark_catalog.default.part [p_partkey#16, p_brand#19, p_size#21, p_container#22] + :- Filter ((((((p_brand#19 == 'Brand#12') and (p_container#22 IN ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG'))) and (p_size#21 >= 1)) and (p_size#21 <= 5)) or ((((p_brand#19 == 'Brand#23') and (p_container#22 IN ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK'))) and (p_size#21 >= 1)) and (p_size#21 <= 10))) or ((((p_brand#19 == 'Brand#34') and (p_container#22 IN ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG'))) and (p_size#21 >= 1)) and (p_size#21 <= 15))) + : +- Scan spark_catalog.default.part [p_partkey#16, p_brand#19, p_size#21, p_container#22] + +- Filter ((((((l_quantity#4 >= 1.00) and (l_quantity#4 <= 11.00)) and (l_shipmode#14 IN ('AIR', 'AIR REG'))) and (l_shipinstruct#13 == 'DELIVER IN PERSON')) or ((((l_quantity#4 >= 10.00) and (l_quantity#4 <= 20.00)) and (l_shipmode#14 IN ('AIR', 'AIR REG'))) and (l_shipinstruct#13 == 'DELIVER IN PERSON'))) or ((((l_quantity#4 >= 20.00) and (l_quantity#4 <= 30.00)) and (l_shipmode#14 IN ('AIR', 'AIR REG'))) and (l_shipinstruct#13 == 'DELIVER IN PERSON'))) + +- Scan spark_catalog.default.lineitem [l_partkey#1, l_quantity#4, l_extendedprice#5, l_discount#6, l_shipinstruct#13, l_shipmode#14] diff --git a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q21.snap b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q21.snap index 92007a4a0..fa8f7c6f1 100644 --- a/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q21.snap +++ b/crates/krishiv-optimizer/tests/it/snapshots/it__tpch__q21.snap @@ -10,20 +10,20 @@ Limit 100 :- Join LeftSemi, ((l_orderkey#36 == l_orderkey#7) and (NOT (l_suppkey#38 == l_suppkey#9))) : :- Project [l_orderkey#7, l_suppkey#9, s_name#1] : : +- Join Inner, (o_orderkey#23 == l_orderkey#7) - : : :- Project [l_orderkey#7, l_suppkey#9, s_name#1] - : : : +- Join Inner, (s_suppkey#0 == l_suppkey#9) - : : : :- Project [l_orderkey#7, l_suppkey#9] - : : : : +- Filter (l_receiptdate#19 > l_commitdate#18) - : : : : +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_commitdate#18, l_receiptdate#19] (shared #1) - : : : +- Project [s_suppkey#0, s_name#1] - : : : +- Join Inner, (s_nationkey#3 == n_nationkey#32) - : : : :- Scan spark_catalog.default.supplier [s_suppkey#0, s_name#1, s_nationkey#3] - : : : +- Project [n_nationkey#32] - : : : +- Filter (n_name#33 == 'SAUDI ARABIA') - : : : +- Scan spark_catalog.default.nation [n_nationkey#32, n_name#33] - : : +- Project [o_orderkey#23] - : : +- Filter (o_orderstatus#25 == 'F') - : : +- Scan spark_catalog.default.orders [o_orderkey#23, o_orderstatus#25] + : : :- Project [o_orderkey#23] + : : : +- Filter (o_orderstatus#25 == 'F') + : : : +- Scan spark_catalog.default.orders [o_orderkey#23, o_orderstatus#25] + : : +- Project [l_orderkey#7, l_suppkey#9, s_name#1] + : : +- Join Inner, (s_suppkey#0 == l_suppkey#9) + : : :- Project [l_orderkey#7, l_suppkey#9] + : : : +- Filter (l_receiptdate#19 > l_commitdate#18) + : : : +- Scan spark_catalog.default.lineitem [l_orderkey#7, l_suppkey#9, l_commitdate#18, l_receiptdate#19] (shared #1) + : : +- Project [s_suppkey#0, s_name#1] + : : +- Join Inner, (s_nationkey#3 == n_nationkey#32) + : : :- Scan spark_catalog.default.supplier [s_suppkey#0, s_name#1, s_nationkey#3] + : : +- Project [n_nationkey#32] + : : +- Filter (n_name#33 == 'SAUDI ARABIA') + : : +- Scan spark_catalog.default.nation [n_nationkey#32, n_name#33] : +- Project [l_orderkey#36, l_suppkey#38] : +- Shift +29 : +- Shared #1, printed above diff --git a/crates/krishiv-optimizer/tests/it/support/mod.rs b/crates/krishiv-optimizer/tests/it/support/mod.rs index f1f5eb87a..2ffb179f4 100644 --- a/crates/krishiv-optimizer/tests/it/support/mod.rs +++ b/crates/krishiv-optimizer/tests/it/support/mod.rs @@ -137,3 +137,33 @@ impl Rng { items[usize::try_from(self.range(0, i64::try_from(items.len()).unwrap() - 1)).unwrap()] } } + +/// Default row counts with every column's distinct count known, all values +/// distinct: the estimates `optimize` made before a literal equality on a +/// column without a distinct count took the default selectivity. Rule tests +/// that need a selective filter on a table without statistics plan under +/// them. +#[derive(Debug)] +pub(crate) struct AllDistinct; + +impl krishiv_optimizer::Statistics for AllDistinct { + fn table(&self, scan: &krishiv_plan::Scan) -> Option { + #[expect( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + reason = "a constant" + )] + let rows = krishiv_optimizer::DEFAULT_ROWS as u64; + let columns = scan.output.iter().map(|c| { + let stats = krishiv_optimizer::ColumnStats { + distinct: Some(rows), + ..krishiv_optimizer::ColumnStats::default() + }; + (c.name.clone(), stats) + }); + Some(krishiv_optimizer::TableStats { + rows, + columns: columns.collect(), + }) + } +} diff --git a/crates/krishiv-plan-codec/src/decode.rs b/crates/krishiv-plan-codec/src/decode.rs index 7d2d94f8c..c7469866c 100644 --- a/crates/krishiv-plan-codec/src/decode.rs +++ b/crates/krishiv-plan-codec/src/decode.rs @@ -841,6 +841,9 @@ fn join_type(t: i32) -> Result { LeftSemi => JoinType::LeftSemi, LeftAnti => JoinType::LeftAnti, NullAwareAnti => JoinType::NullAwareAnti, + 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 db5e26505..f8d0cceb6 100644 --- a/crates/krishiv-plan-codec/src/encode.rs +++ b/crates/krishiv-plan-codec/src/encode.rs @@ -532,6 +532,9 @@ fn join_type(t: JoinType) -> pb::JoinType { JoinType::LeftSemi => pb::JoinType::LeftSemi, JoinType::LeftAnti => pb::JoinType::LeftAnti, JoinType::NullAwareAnti => pb::JoinType::NullAwareAnti, + 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 87a68bca4..af80ab004 100644 --- a/crates/krishiv-plan-codec/src/features.rs +++ b/crates/krishiv-plan-codec/src/features.rs @@ -115,6 +115,7 @@ macro_rules! registry { // (id, variant, legacy level, kind, meaning) registry! { (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`)."), (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_group_limit, WindowGroupLimit, POST_MIGRATION_READER_VERSION, Optimization, "The `WindowGroupLimit` node (the rows of each partition within a rank)."), @@ -188,6 +189,9 @@ pub fn feature_of(construct: Construct<'_>) -> Option { | JoinType::LeftSemi | JoinType::LeftAnti | JoinType::NullAwareAnti => None, + JoinType::RightSemi | JoinType::RightAnti | JoinType::RightOuter => { + Some(Feature::JoinBuildPreserved) + } }, Construct::Partitioning(p) => match p { Partitioning::Hash { .. } 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 1318b5f58..5962bfb41 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 @@ -1124,6 +1124,14 @@ pub enum JoinType { LeftSemi = 3, LeftAnti = 4, NullAwareAnti = 5, + /// Joins that build their preserved side and return its rows by their + /// matches. + /// feature: join_build_preserved + RightSemi = 6, + /// feature: join_build_preserved + RightAnti = 7, + /// feature: join_build_preserved + RightOuter = 8, } impl JoinType { /// String value of the enum field names used in the ProtoBuf definition. @@ -1138,6 +1146,9 @@ impl JoinType { Self::LeftSemi => "JOIN_TYPE_LEFT_SEMI", Self::LeftAnti => "JOIN_TYPE_LEFT_ANTI", Self::NullAwareAnti => "JOIN_TYPE_NULL_AWARE_ANTI", + Self::RightSemi => "JOIN_TYPE_RIGHT_SEMI", + Self::RightAnti => "JOIN_TYPE_RIGHT_ANTI", + Self::RightOuter => "JOIN_TYPE_RIGHT_OUTER", } } /// Creates an enum from field names used in the ProtoBuf definition. @@ -1149,6 +1160,9 @@ impl JoinType { "JOIN_TYPE_LEFT_SEMI" => Some(Self::LeftSemi), "JOIN_TYPE_LEFT_ANTI" => Some(Self::LeftAnti), "JOIN_TYPE_NULL_AWARE_ANTI" => Some(Self::NullAwareAnti), + "JOIN_TYPE_RIGHT_SEMI" => Some(Self::RightSemi), + "JOIN_TYPE_RIGHT_ANTI" => Some(Self::RightAnti), + "JOIN_TYPE_RIGHT_OUTER" => Some(Self::RightOuter), _ => None, } } diff --git a/crates/krishiv-plan-codec/tests/common/inventory.rs b/crates/krishiv-plan-codec/tests/common/inventory.rs index 86d9097d2..c42eb2ec9 100644 --- a/crates/krishiv-plan-codec/tests/common/inventory.rs +++ b/crates/krishiv-plan-codec/tests/common/inventory.rs @@ -532,6 +532,27 @@ pub(crate) fn nodes() -> Vec<(&'static str, PhysNode)> { ..join(JoinType::NullAwareAnti) }), ), + ( + "hash_join_right_semi", + PhysNode::HashJoin(HashJoinPlan { + distribution: JoinDistribution::Partitioned, + ..join(JoinType::RightSemi) + }), + ), + ( + "hash_join_right_anti", + PhysNode::HashJoin(HashJoinPlan { + distribution: JoinDistribution::Partitioned, + ..join(JoinType::RightAnti) + }), + ), + ( + "hash_join_right_outer", + PhysNode::HashJoin(HashJoinPlan { + distribution: JoinDistribution::Partitioned, + ..join(JoinType::RightOuter) + }), + ), ( "hash_join_partitioned", PhysNode::HashJoin(HashJoinPlan { diff --git a/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_anti.bin b/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_anti.bin new file mode 100644 index 000000000..e99603ece Binary files /dev/null and b/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_anti.bin differ diff --git a/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_outer.bin b/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_outer.bin new file mode 100644 index 000000000..f5bd38c5b Binary files /dev/null and b/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_outer.bin differ diff --git a/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_semi.bin b/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_semi.bin new file mode 100644 index 000000000..5501fde4a Binary files /dev/null and b/crates/krishiv-plan-codec/tests/golden/join_build_preserved/hash_join_right_semi.bin differ diff --git a/crates/krishiv-plan-codec/tests/roundtrip.rs b/crates/krishiv-plan-codec/tests/roundtrip.rs index b69a5435f..e48fcbd68 100644 --- a/crates/krishiv-plan-codec/tests/roundtrip.rs +++ b/crates/krishiv-plan-codec/tests/roundtrip.rs @@ -15,8 +15,8 @@ use krishiv_exec::physical::{PhysNode, PhysicalPlan}; use krishiv_plan::{Expr, ReadSource}; use krishiv_plan_codec::proto as pb; use krishiv_plan_codec::{ - CodecError, FORMAT_VERSION, Limits, MIN_READER_VERSION, ReaderFeatures, decode, decode_with, - encode, to_message, + CodecError, FORMAT_VERSION, Feature, FeatureKind, Limits, MIN_READER_VERSION, + POST_MIGRATION_READER_VERSION, ReaderFeatures, decode, decode_with, encode, to_message, }; use prost::Message as _; @@ -294,6 +294,7 @@ fn readers_that_predate_features_still_refuse_what_they_cannot_read() { n if n.starts_with("expand") => 5, // 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, _ => MIN_READER_VERSION, }; assert_eq!(m.min_reader_version, expected, "{name}"); @@ -433,6 +434,46 @@ fn a_corrupt_expand_is_refused() { ); } +#[test] +fn joins_that_build_their_preserved_side_are_a_feature() { + // The three join types are the `join_build_preserved` feature, which + // the planner can avoid (it builds the other side instead). A reader + // that predates features refuses the plan as too new rather than as an + // unknown join type. + assert_eq!( + Feature::JoinBuildPreserved.kind(), + FeatureKind::Optimization + ); + for name in ["hash_join", "hash_join_anti", "nested_loop_join"] { + let m = message(name); + assert!(m.features.is_empty(), "{name}"); + assert_eq!(m.min_reader_version, MIN_READER_VERSION, "{name}"); + } + for name in [ + "hash_join_right_semi", + "hash_join_right_anti", + "hash_join_right_outer", + ] { + let m = message(name); + assert_eq!(m.features, ["join_build_preserved"], "{name}"); + assert_eq!( + m.min_reader_version, POST_MIGRATION_READER_VERSION, + "{name}" + ); + } +} + +#[test] +fn an_unspecified_join_type_is_corrupt() { + let mut m = message("hash_join"); + for node in &mut m.nodes { + if let Some(pb::node::Kind::HashJoin(j)) = &mut node.kind { + j.join_type = pb::JoinType::Unspecified as i32; + } + } + corrupt(&m, "join type"); +} + #[test] fn a_newer_writer_that_stays_compatible_is_read() { // A writer that only added ignorable fields: unknown fields (here diff --git a/proto/krishiv/plan/v1/plan.proto b/proto/krishiv/plan/v1/plan.proto index 988ffc955..9fa38e5bb 100644 --- a/proto/krishiv/plan/v1/plan.proto +++ b/proto/krishiv/plan/v1/plan.proto @@ -333,6 +333,14 @@ enum JoinType { JOIN_TYPE_LEFT_SEMI = 3; JOIN_TYPE_LEFT_ANTI = 4; JOIN_TYPE_NULL_AWARE_ANTI = 5; + // Joins that build their preserved side and return its rows by their + // matches. + // feature: join_build_preserved + JOIN_TYPE_RIGHT_SEMI = 6; + // feature: join_build_preserved + JOIN_TYPE_RIGHT_ANTI = 7; + // feature: join_build_preserved + JOIN_TYPE_RIGHT_OUTER = 8; } message Residual {