Join planning: literal-equality selectivity without distinct counts; join types that build their preserved side; build-side preference

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());

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<Self> {
         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<Field> {
             spec.probe_schema
                 .fields()

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<Option<Batch>> {
         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,
         };

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<LogicalPlan>) -> 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<LogicalPlan>,

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<Arc<Expr>> = 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<R
 }
 
 impl Planner<'_> {
+    /// The statistics of a join input planned as `schema` from `plan`.
+    fn input_stats(&self, plan: &Arc<LogicalPlan>, 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(

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;