Prototype for Join Ordering

datafusion/catalog/src/default_table_source.rs

@@ -83,6 +83,10 @@ impl TableSource for DefaultTableSource {
     fn get_column_default(&self, column: &str) -> Option<&Expr> {
         self.table_provider.get_column_default(column)
     }
+
+    fn statistics(&self) -> Option<datafusion_common::Statistics> {
+        self.table_provider.statistics()
+    }
 }
 
 /// Wrap TableProvider in TableSource

datafusion/expr/src/table_source.rs

@@ -20,7 +20,7 @@
 use crate::{Expr, LogicalPlan};
 
 use arrow::datatypes::SchemaRef;
-use datafusion_common::{Constraints, Result};
+use datafusion_common::{Constraints, Result, Statistics};
 
 use std::{any::Any, borrow::Cow};
 
@@ -129,4 +129,12 @@ pub trait TableSource: Sync + Send {
     fn get_column_default(&self, _column: &str) -> Option<&Expr> {
         None
     }
+
+    /// Get statistics for this table source, if available
+    /// Although not presently used in mainline DataFusion, this allows implementation specific
+    /// behavior for downstream repositories, in conjunction with specialized optimizer rules to
+    /// perform operations such as re-ordering of joins.
+    fn statistics(&self) -> Option<Statistics> {
+        None
+    }
 }

datafusion/optimizer/src/lib.rs

@@ -58,6 +58,7 @@ pub mod optimizer;
 pub mod propagate_empty_relation;
 pub mod push_down_filter;
 pub mod push_down_limit;
+pub mod reorder_join;
 pub mod replace_distinct_aggregate;
 pub mod scalar_subquery_to_join;
 pub mod simplify_expressions;

datafusion/optimizer/src/reorder_join/cost.rs

@@ -0,0 +1,59 @@
+use datafusion_common::{plan_datafusion_err, plan_err, stats::Precision, Result};
+use datafusion_expr::{Join, JoinType, LogicalPlan};
+
+pub trait JoinCostEstimator: std::fmt::Debug {
+    fn cardinality(&self, plan: &LogicalPlan) -> Option<f64> {
+        estimate_cardinality(plan).ok()
+    }
+
+    fn selectivity(&self, join: &Join) -> f64 {
+        match join.join_type {
+            JoinType::Inner => 0.1,
+            _ => 1.0,
+        }
+    }
+
+    fn cost(&self, selectivity: f64, cardinality: f64) -> f64 {
+        selectivity * cardinality
+    }
+}
+
+/// Default implementation of JoinCostEstimator
+#[derive(Debug, Clone, Copy)]
+pub struct DefaultCostEstimator;
+
+impl JoinCostEstimator for DefaultCostEstimator {}
+
+fn estimate_cardinality(plan: &LogicalPlan) -> Result<f64> {
+    match plan {
+        LogicalPlan::Filter(filter) => {
+            let input_cardinality = estimate_cardinality(&filter.input)?;
+            Ok(0.1 * input_cardinality)
+        }
+        LogicalPlan::Aggregate(agg) => {
+            let input_cardinality = estimate_cardinality(&agg.input)?;
+            Ok(0.1 * input_cardinality)
+        }
+        LogicalPlan::TableScan(scan) => {
+            let statistics = scan
+                .source
+                .statistics()
+                .ok_or_else(|| plan_datafusion_err!("Table statistics not available"))?;
+            if let Precision::Exact(num_rows) | Precision::Inexact(num_rows) =
+                statistics.num_rows
+            {
+                Ok(num_rows as f64)
+            } else {
+                plan_err!("Number of rows not available")
+            }
+        }
+        x => {
+            let inputs = x.inputs();
+            if inputs.len() == 1 {
+                estimate_cardinality(inputs[0])
+            } else {
+                plan_err!("Cannot estimate cardinality for plan with multiple inputs")
+            }
+        }
+    }
+}