refactor(df-repr): remove cross join, use inner join (#265)

Signed-off-by: Alex Chi Z <iskyzh@gmail.com>

Author: skyzh

optd-datafusion-bridge/src/from_optd.rs

@@ -476,7 +476,9 @@ impl OptdPlanContext<'_> {
         let physical_expr =
             self.conv_from_optd_expr(node.cond(), &Arc::new(filter_schema.clone()))?;
 
-        if *node.join_type() == JoinType::Cross {
+        if node.join_type() == &JoinType::Inner
+            && node.cond() == ConstantPred::bool(true).into_pred_node()
+        {
             return Ok(Arc::new(CrossJoinExec::new(left_exec, right_exec))
                 as Arc<dyn ExecutionPlan + 'static>);
         }
@@ -491,7 +493,6 @@ impl OptdPlanContext<'_> {
             JoinType::LeftAnti => datafusion_expr::JoinType::LeftAnti,
             JoinType::RightAnti => datafusion_expr::JoinType::RightAnti,
             JoinType::LeftMark => datafusion_expr::JoinType::LeftMark,
-            _ => unimplemented!(),
         };
 
         let mut column_idxs = vec![];

optd-datafusion-bridge/src/into_optd.rs

@@ -547,7 +547,7 @@ impl OptdPlanContext<'_> {
                 left,
                 right,
                 ConstantPred::bool(true).into_pred_node(),
-                JoinType::Cross,
+                JoinType::Inner,
             ))
         } else if log_ops.len() == 1 {
             Ok(LogicalJoin::new(left, right, log_ops.remove(0), join_type))

optd-datafusion-repr-adv-cost/src/adv_stats/join.rs

@@ -191,13 +191,6 @@ impl<
             JoinType::Inner => inner_join_selectivity,
             JoinType::LeftOuter => f64::max(inner_join_selectivity, 1.0 / right_row_cnt),
             JoinType::RightOuter => f64::max(inner_join_selectivity, 1.0 / left_row_cnt),
-            JoinType::Cross => {
-                assert!(
-                    on_col_ref_pairs.is_empty(),
-                    "Cross joins should not have on columns"
-                );
-                join_filter_selectivity
-            }
             // TODO: Does this make sense?
             JoinType::LeftMark => f64::max(inner_join_selectivity, 1.0 / right_row_cnt),
             _ => unimplemented!("join_typ={} is not implemented", join_typ),

optd-datafusion-repr/src/plan_nodes/join.rs

@@ -15,7 +15,6 @@ pub enum JoinType {
     FullOuter,
     LeftOuter,
     RightOuter,
-    Cross,
     LeftSemi,
     RightSemi,
     LeftAnti,

optd-datafusion-repr/src/properties/column_ref.rs

@@ -471,7 +471,7 @@ impl LogicalPropertyBuilder<DfNodeType> for ColumnRefPropertyBuilder {
                 //
                 // Otherwise be conservative and discard all correlations.
                 let output_correlation = match join_type {
-                    JoinType::Inner | JoinType::Cross => {
+                    JoinType::Inner => {
                         // Merge the equal columns in the join condition into the those from the
                         // children.
                         if let Some(SemanticCorrelation {

optd-datafusion-repr/src/properties/schema.rs

@@ -180,7 +180,7 @@ impl LogicalPropertyBuilder<DfNodeType> for SchemaPropertyBuilder {
             DfNodeType::Join(join_type) => {
                 use crate::plan_nodes::JoinType::*;
                 match join_type {
-                    Inner | LeftOuter | RightOuter | FullOuter | Cross => {
+                    Inner | LeftOuter | RightOuter | FullOuter => {
                         let mut schema = children[0].clone();
                         let schema2 = children[1].clone();
                         schema.fields.extend(schema2.fields);

optd-datafusion-repr/src/rules/filter_pushdown.rs

@@ -160,20 +160,6 @@ fn apply_filter_merge(
     vec![new_filter.into_plan_node().into()]
 }
 
-// TODO: define_rule! should be able to match on any join type, ideally...
-define_rule!(
-    FilterCrossJoinTransposeRule,
-    apply_filter_cross_join_transpose,
-    (Filter, (Join(JoinType::Cross), child_a, child_b))
-);
-
-fn apply_filter_cross_join_transpose(
-    optimizer: &impl Optimizer<DfNodeType>,
-    binding: ArcDfPlanNode,
-) -> Vec<PlanNodeOrGroup<DfNodeType>> {
-    filter_join_transpose(optimizer, binding)
-}
-
 define_rule!(
     FilterInnerJoinTransposeRule,
     apply_filter_inner_join_transpose,
@@ -256,18 +242,6 @@ fn filter_join_transpose(
             let new_conds = merge_conds(and_expr_list_to_expr(join_conds), old_cond);
             LogicalJoin::new_unchecked(new_left, new_right, new_conds, JoinType::Inner)
         }
-        JoinType::Cross => {
-            if !join_conds.is_empty() {
-                LogicalJoin::new_unchecked(
-                    new_left,
-                    new_right,
-                    and_expr_list_to_expr(join_conds),
-                    JoinType::Inner,
-                )
-            } else {
-                LogicalJoin::new_unchecked(new_left, new_right, join_cond, JoinType::Cross)
-            }
-        }
         _ => {
             // We don't support modifying the join condition for other join types yet
             LogicalJoin::new_unchecked(new_left, new_right, join_cond, *join_typ)

optd-datafusion-repr/src/rules/joins.rs

@@ -81,7 +81,7 @@ fn apply_eliminate_join(
                         left,
                         right,
                         ConstantPred::bool(true).into_pred_node(),
-                        JoinType::Cross,
+                        JoinType::Inner,
                     );
                     return vec![node.into_plan_node().into()];
                 } else {

optd-datafusion-repr/src/rules/subquery/depjoin_pushdown.rs

@@ -86,7 +86,7 @@ fn apply_dep_initial_distinct(
                 left,
                 right,
                 ConstantPred::bool(true).into_pred_node(),
-                JoinType::Cross,
+                JoinType::Inner,
             )
             .into_plan_node(),
             SubqueryType::Exists => {
@@ -122,7 +122,7 @@ fn apply_dep_initial_distinct(
                     left,
                     count_star_to_bool_proj,
                     ConstantPred::bool(true).into_pred_node(),
-                    JoinType::Cross,
+                    JoinType::Inner,
                 )
                 .into_plan_node()
             }
@@ -147,7 +147,7 @@ fn apply_dep_initial_distinct(
             left,
             right,
             ConstantPred::bool(true).into_pred_node(),
-            JoinType::Cross,
+            JoinType::Inner,
         );
 
         return vec![new_join.into_plan_node().into()];