Feat: flatten filter and join cond to list of expr, simplify filter rule (#108)

# Major Changes
- BinOpExpr in Join cond and filter cond is changed to LogOpExpr
- LogOpExpr child is not a single ExprList but multiple exprs as
children (which is different from previous logOpExpr)
- ColumnRef and Schema properties changed for LogOpExpr

# Design
Previously, LogOpExpr is designed to be `children: vec![ExprList]` which
means the LogOpExpr only has one child of ExprList. And condition
expressions in filter and join from datafusion will be converted to
BinOpExpr, which means the LogOpExpr actually does not exist in optd.
After converting conditions from datafusion to be LogOpExpr, the design
of one child LogOpExpr will influence the ProjectPullUpJoinRule and
other rules with schema recalculated as these rules simply traverse the
expr's children in its relNode format, it will treat LogOpExpr children
as only one child (ExprList), and when it further traverses ExprList it
has no idea whether it is a LogOp list or a normal list.

So current design change the LogOpExpr to include multiple children to
represent its real children size in RelNode format.
It has further impact on cost model and properties builder logic as
previously List is the only type in Optd which contains multiple
children.

About the pickMany stuff, it won't change the core's logic no matter it has one or many children as we only uses as join cond or filter cond, their cond is only one expr which is LogOpExpr and LogOpExpr is expanded and returned with all of it children. So the LogOpExpr already play the role as a one child wrapper(like ExprList), no need to add a ExprList.

#TODO
- [x] currently I simply copy properties derive and build rules for List
as the rules for LogOpExpr, please check whether it is aligned with cost
model's logic
- [x] the change break the selectivity infer rules and cost model
calculation so it cannot find a best group bindings for several cases.
Please update the selectivity calculation for logOpExpr.

---------

Signed-off-by: AveryQi115 <[email protected]>
Co-authored-by: Patrick Wang <[email protected]>

Author: AveryQi115

datafusion-optd-cli/tpch-sf0_01/simple_manual_test.sql

@@ -10,4 +10,4 @@ CREATE TABLE NATION  (
 CREATE EXTERNAL TABLE nation_tbl STORED AS CSV DELIMITER '|' LOCATION 'datafusion-optd-cli/tpch-sf0_01/nation.tbl';
 insert into nation select column_1, column_2, column_3, column_4 from nation_tbl;
 
-SELECT * FROM nation where nation.n_nationkey = 1;
+SELECT * FROM nation where nation.n_nationkey = 1 OR nation.n_nationkey = 2 OR nation.n_nationkey = 5;

optd-datafusion-bridge/src/from_optd.rs

@@ -25,10 +25,10 @@ use optd_core::rel_node::RelNodeMetaMap;
 use optd_datafusion_repr::{
     plan_nodes::{
         BetweenExpr, BinOpExpr, BinOpType, CastExpr, ColumnRefExpr, ConstantExpr, ConstantType,
-        Expr, FuncExpr, FuncType, InListExpr, JoinType, LikeExpr, OptRelNode, OptRelNodeRef,
-        OptRelNodeTyp, PhysicalAgg, PhysicalEmptyRelation, PhysicalFilter, PhysicalHashJoin,
-        PhysicalLimit, PhysicalNestedLoopJoin, PhysicalProjection, PhysicalScan, PhysicalSort,
-        PlanNode, SortOrderExpr, SortOrderType,
+        Expr, ExprList, FuncExpr, FuncType, InListExpr, JoinType, LikeExpr, LogOpExpr, LogOpType,
+        OptRelNode, OptRelNodeRef, OptRelNodeTyp, PhysicalAgg, PhysicalEmptyRelation,
+        PhysicalFilter, PhysicalHashJoin, PhysicalLimit, PhysicalNestedLoopJoin,
+        PhysicalProjection, PhysicalScan, PhysicalSort, PlanNode, SortOrderExpr, SortOrderType,
     },
     properties::schema::Schema as OptdSchema,
 };
@@ -196,6 +196,23 @@ impl OptdPlanContext<'_> {
                 }
             }
             OptRelNodeTyp::Sort => unreachable!(),
+            OptRelNodeTyp::LogOp(typ) => {
+                let expr = LogOpExpr::from_rel_node(expr.into_rel_node()).unwrap();
+                let mut children = expr.children().into_iter();
+                let first_expr = Self::conv_from_optd_expr(children.next().unwrap(), context)?;
+                let op = match typ {
+                    LogOpType::And => Operator::And,
+                    LogOpType::Or => Operator::Or,
+                };
+                children.try_fold(first_expr, |acc, expr| {
+                    let expr = Self::conv_from_optd_expr(expr, context)?;
+                    Ok(
+                        Arc::new(datafusion::physical_plan::expressions::BinaryExpr::new(
+                            acc, op, expr,
+                        )) as Arc<dyn PhysicalExpr>,
+                    )
+                })
+            }
             OptRelNodeTyp::BinOp(op) => {
                 let expr = BinOpExpr::from_rel_node(expr.into_rel_node()).unwrap();
                 let left = Self::conv_from_optd_expr(expr.left_child(), context)?;
@@ -207,8 +224,6 @@ impl OptdPlanContext<'_> {
                     BinOpType::Lt => Operator::Lt,
                     BinOpType::Geq => Operator::GtEq,
                     BinOpType::Gt => Operator::Gt,
-                    BinOpType::And => Operator::And,
-                    BinOpType::Or => Operator::Or,
                     BinOpType::Add => Operator::Plus,
                     BinOpType::Sub => Operator::Minus,
                     BinOpType::Mul => Operator::Multiply,
@@ -225,10 +240,12 @@ impl OptdPlanContext<'_> {
                 // TODO: should we just convert between to x <= c1 and x >= c2?
                 let expr = BetweenExpr::from_rel_node(expr.into_rel_node()).unwrap();
                 Self::conv_from_optd_expr(
-                    BinOpExpr::new(
-                        BinOpExpr::new(expr.child(), expr.lower(), BinOpType::Geq).into_expr(),
-                        BinOpExpr::new(expr.child(), expr.upper(), BinOpType::Leq).into_expr(),
-                        BinOpType::And,
+                    LogOpExpr::new(
+                        LogOpType::And,
+                        ExprList::new(vec![
+                            BinOpExpr::new(expr.child(), expr.lower(), BinOpType::Geq).into_expr(),
+                            BinOpExpr::new(expr.child(), expr.upper(), BinOpType::Leq).into_expr(),
+                        ]),
                     )
                     .into_expr(),
                     context,

optd-datafusion-bridge/src/into_optd.rs

@@ -8,13 +8,35 @@ use datafusion_expr::Expr as DFExpr;
 use optd_core::rel_node::RelNode;
 use optd_datafusion_repr::plan_nodes::{
     BetweenExpr, BinOpExpr, BinOpType, CastExpr, ColumnRefExpr, ConstantExpr, Expr, ExprList,
-    FuncExpr, FuncType, InListExpr, JoinType, LikeExpr, LogicalAgg, LogicalEmptyRelation,
-    LogicalFilter, LogicalJoin, LogicalLimit, LogicalProjection, LogicalScan, LogicalSort,
-    OptRelNode, OptRelNodeRef, OptRelNodeTyp, PlanNode, SortOrderExpr, SortOrderType,
+    FuncExpr, FuncType, InListExpr, JoinType, LikeExpr, LogOpExpr, LogOpType, LogicalAgg,
+    LogicalEmptyRelation, LogicalFilter, LogicalJoin, LogicalLimit, LogicalProjection, LogicalScan,
+    LogicalSort, OptRelNode, OptRelNodeRef, OptRelNodeTyp, PlanNode, SortOrderExpr, SortOrderType,
 };
 
 use crate::OptdPlanContext;
 
+// flatten_nested_logical is a helper function to flatten nested logical operators with same op type
+// eg. (a AND (b AND c)) => ExprList([a, b, c])
+//    (a OR (b OR c)) => ExprList([a, b, c])
+// It assume the children of the input expr_list are already flattened
+//  and can only be used in bottom up manner
+fn flatten_nested_logical(op: LogOpType, expr_list: ExprList) -> ExprList {
+    // conv_into_optd_expr is building the children bottom up so there is no need to
+    // call flatten_nested_logical recursively
+    let mut new_expr_list = Vec::new();
+    for child in expr_list.to_vec() {
+        if let OptRelNodeTyp::LogOp(child_op) = child.typ() {
+            if child_op == op {
+                let child_log_op_expr = LogOpExpr::from_rel_node(child.into_rel_node()).unwrap();
+                new_expr_list.extend(child_log_op_expr.children().to_vec());
+                continue;
+            }
+        }
+        new_expr_list.push(child.clone());
+    }
+    ExprList::new(new_expr_list)
+}
+
 impl OptdPlanContext<'_> {
     fn conv_into_optd_table_scan(&mut self, node: &logical_plan::TableScan) -> Result<PlanNode> {
         let table_name = node.table_name.to_string();
@@ -47,15 +69,29 @@ impl OptdPlanContext<'_> {
             Expr::BinaryExpr(node) => {
                 let left = self.conv_into_optd_expr(node.left.as_ref(), context)?;
                 let right = self.conv_into_optd_expr(node.right.as_ref(), context)?;
+                match node.op {
+                    Operator::And => {
+                        let op = LogOpType::And;
+                        let expr_list = ExprList::new(vec![left, right]);
+                        let expr_list = flatten_nested_logical(op, expr_list);
+                        return Ok(LogOpExpr::new(op, expr_list).into_expr());
+                    }
+                    Operator::Or => {
+                        let op = LogOpType::Or;
+                        let expr_list = ExprList::new(vec![left, right]);
+                        let expr_list = flatten_nested_logical(op, expr_list);
+                        return Ok(LogOpExpr::new(op, expr_list).into_expr());
+                    }
+                    _ => {}
+                }
+
                 let op = match node.op {
                     Operator::Eq => BinOpType::Eq,
                     Operator::NotEq => BinOpType::Neq,
                     Operator::LtEq => BinOpType::Leq,
                     Operator::Lt => BinOpType::Lt,
                     Operator::GtEq => BinOpType::Geq,
                     Operator::Gt => BinOpType::Gt,
-                    Operator::And => BinOpType::And,
-                    Operator::Or => BinOpType::Or,
                     Operator::Plus => BinOpType::Add,
                     Operator::Minus => BinOpType::Sub,
                     Operator::Multiply => BinOpType::Mul,
@@ -305,19 +341,13 @@ impl OptdPlanContext<'_> {
         } else if log_ops.len() == 1 {
             Ok(LogicalJoin::new(left, right, log_ops.remove(0), join_type))
         } else {
-            // Build a left-deep tree from log_ops
-            // I wanted to pop from the left instead of the right to maintain the order, even if it's slower
-            // you can obv change log_ops to a Deque to avoid this issue but I didn't bother since I don't wanna
-            // do premature optimization
-            let left_nonlog_op = log_ops.remove(0);
-            let right_nonlog_op = log_ops.remove(0);
-            let mut cond =
-                BinOpExpr::new(left_nonlog_op, right_nonlog_op, BinOpType::And).into_expr();
-            while !log_ops.is_empty() {
-                cond = BinOpExpr::new(cond, log_ops.remove(0), BinOpType::And).into_expr();
-            }
-
-            Ok(LogicalJoin::new(left, right, cond, join_type))
+            let expr_list = ExprList::new(log_ops);
+            Ok(LogicalJoin::new(
+                left,
+                right,
+                LogOpExpr::new(LogOpType::And, expr_list).into_expr(),
+                join_type,
+            ))
         }
     }
 

optd-datafusion-repr/src/cost/base_cost.rs

@@ -1,6 +1,6 @@
 use std::{collections::HashMap, sync::Arc};
 
-use crate::plan_nodes::{BinOpType, ColumnRefExpr, OptRelNode, UnOpType};
+use crate::plan_nodes::{BinOpType, ColumnRefExpr, LogOpType, OptRelNode, UnOpType};
 use crate::properties::column_ref::{ColumnRefPropertyBuilder, GroupColumnRefs};
 use crate::{
     plan_nodes::{OptRelNodeRef, OptRelNodeTyp},
@@ -298,19 +298,15 @@ impl OptCostModel {
                         right_child,
                         column_refs,
                     )
-                } else if bin_op_typ.is_logical() {
-                    self.get_logical_bin_op_selectivity(
-                        bin_op_typ,
-                        left_child,
-                        right_child,
-                        column_refs,
-                    )
                 } else if bin_op_typ.is_numerical() {
                     INVALID_SELECTIVITY
                 } else {
                     unreachable!("all BinOpTypes should be true for at least one is_*() function")
                 }
             }
+            OptRelNodeTyp::LogOp(log_op_typ) => {
+                self.get_log_op_selectivity(log_op_typ, &expr_tree.children, column_refs)
+            }
             OptRelNodeTyp::UnOp(un_op_typ) => {
                 assert!(expr_tree.children.len() == 1);
                 let child = expr_tree.child(0);
@@ -532,24 +528,20 @@ impl OptCostModel {
         }
     }
 
-    fn get_logical_bin_op_selectivity(
+    fn get_log_op_selectivity(
         &self,
-        bin_op_typ: BinOpType,
-        left: OptRelNodeRef,
-        right: OptRelNodeRef,
+        log_op_typ: LogOpType,
+        children: &[OptRelNodeRef],
         column_refs: &GroupColumnRefs,
     ) -> f64 {
-        assert!(bin_op_typ.is_logical());
-
-        let left_sel = self.get_filter_selectivity(left, column_refs);
-        let right_sel = self.get_filter_selectivity(right, column_refs);
-
-        match bin_op_typ {
-            // note that there's no need to account for nulls here
-            // it's also impossible to even account for nulls because we don't know which columns the left and right selectivities are
-            BinOpType::And => left_sel * right_sel,
-            BinOpType::Or => left_sel + right_sel - left_sel * right_sel,
-            _ => unreachable!("we covered all bin_op_typ.is_logical() cases"),
+        let children_sel = children
+            .iter()
+            .map(|expr| self.get_filter_selectivity(expr.clone(), column_refs));
+
+        match log_op_typ {
+            LogOpType::And => children_sel.product(),
+            // the formula is 1.0 - the probability of _none_ of the events happening
+            LogOpType::Or => 1.0 - children_sel.fold(1.0, |acc, sel| acc * (1.0 - sel)),
         }
     }
 
@@ -595,8 +587,8 @@ mod tests {
 
     use crate::{
         plan_nodes::{
-            BinOpExpr, BinOpType, ColumnRefExpr, ConstantExpr, Expr, OptRelNode, OptRelNodeRef,
-            UnOpExpr, UnOpType,
+            BinOpExpr, BinOpType, ColumnRefExpr, ConstantExpr, Expr, ExprList, LogOpExpr,
+            LogOpType, OptRelNode, OptRelNodeRef, UnOpExpr, UnOpType,
         },
         properties::column_ref::ColumnRef,
     };
@@ -696,6 +688,21 @@ mod tests {
         .into_rel_node()
     }
 
+    fn log_op(op_type: LogOpType, children: Vec<OptRelNodeRef>) -> OptRelNodeRef {
+        LogOpExpr::new(
+            op_type,
+            ExprList::new(
+                children
+                    .into_iter()
+                    .map(|opt_rel_node_ref| {
+                        Expr::from_rel_node(opt_rel_node_ref).expect("all children should be Expr")
+                    })
+                    .collect(),
+            ),
+        )
+        .into_rel_node()
+    }
+
     fn un_op(op_type: UnOpType, child: OptRelNodeRef) -> OptRelNodeRef {
         UnOpExpr::new(
             Expr::from_rel_node(child).expect("child should be an Expr"),
@@ -1136,59 +1143,79 @@ mod tests {
     fn test_and() {
         let cost_model = create_one_column_cost_model(PerColumnStats::new(
             Box::new(MockMostCommonValues {
-                mcvs: vec![(Value::Int32(1), 0.3), (Value::Int32(5), 0.5)]
-                    .into_iter()
-                    .collect(),
+                mcvs: vec![
+                    (Value::Int32(1), 0.3),
+                    (Value::Int32(5), 0.5),
+                    (Value::Int32(8), 0.2),
+                ]
+                .into_iter()
+                .collect(),
             }),
             0,
             0.0,
             Box::new(MockDistribution::empty()),
         ));
         let eq1 = bin_op(BinOpType::Eq, col_ref(0), cnst(Value::Int32(1)));
         let eq5 = bin_op(BinOpType::Eq, col_ref(0), cnst(Value::Int32(5)));
-        let expr_tree = bin_op(BinOpType::And, eq1.clone(), eq5.clone());
-        let expr_tree_rev = bin_op(BinOpType::And, eq5.clone(), eq1.clone());
+        let eq8 = bin_op(BinOpType::Eq, col_ref(0), cnst(Value::Int32(8)));
+        let expr_tree = log_op(LogOpType::And, vec![eq1.clone(), eq5.clone(), eq8.clone()]);
+        let expr_tree_shift1 = log_op(LogOpType::And, vec![eq5.clone(), eq8.clone(), eq1.clone()]);
+        let expr_tree_shift2 = log_op(LogOpType::And, vec![eq8.clone(), eq1.clone(), eq5.clone()]);
         let column_refs = vec![ColumnRef::BaseTableColumnRef {
             table: String::from(TABLE1_NAME),
             col_idx: 0,
         }];
         assert_approx_eq::assert_approx_eq!(
             cost_model.get_filter_selectivity(expr_tree, &column_refs),
-            0.15
+            0.03
         );
         assert_approx_eq::assert_approx_eq!(
-            cost_model.get_filter_selectivity(expr_tree_rev, &column_refs),
-            0.15
+            cost_model.get_filter_selectivity(expr_tree_shift1, &column_refs),
+            0.03
+        );
+        assert_approx_eq::assert_approx_eq!(
+            cost_model.get_filter_selectivity(expr_tree_shift2, &column_refs),
+            0.03
         );
     }
 
     #[test]
     fn test_or() {
         let cost_model = create_one_column_cost_model(PerColumnStats::new(
             Box::new(MockMostCommonValues {
-                mcvs: vec![(Value::Int32(1), 0.3), (Value::Int32(5), 0.5)]
-                    .into_iter()
-                    .collect(),
+                mcvs: vec![
+                    (Value::Int32(1), 0.3),
+                    (Value::Int32(5), 0.5),
+                    (Value::Int32(8), 0.2),
+                ]
+                .into_iter()
+                .collect(),
             }),
             0,
             0.0,
             Box::new(MockDistribution::empty()),
         ));
         let eq1 = bin_op(BinOpType::Eq, col_ref(0), cnst(Value::Int32(1)));
         let eq5 = bin_op(BinOpType::Eq, col_ref(0), cnst(Value::Int32(5)));
-        let expr_tree = bin_op(BinOpType::Or, eq1.clone(), eq5.clone());
-        let expr_tree_rev = bin_op(BinOpType::Or, eq5.clone(), eq1.clone());
+        let eq8 = bin_op(BinOpType::Eq, col_ref(0), cnst(Value::Int32(8)));
+        let expr_tree = log_op(LogOpType::Or, vec![eq1.clone(), eq5.clone(), eq8.clone()]);
+        let expr_tree_shift1 = log_op(LogOpType::Or, vec![eq5.clone(), eq8.clone(), eq1.clone()]);
+        let expr_tree_shift2 = log_op(LogOpType::Or, vec![eq8.clone(), eq1.clone(), eq5.clone()]);
         let column_refs = vec![ColumnRef::BaseTableColumnRef {
             table: String::from(TABLE1_NAME),
             col_idx: 0,
         }];
         assert_approx_eq::assert_approx_eq!(
             cost_model.get_filter_selectivity(expr_tree, &column_refs),
-            0.65
+            0.72
         );
         assert_approx_eq::assert_approx_eq!(
-            cost_model.get_filter_selectivity(expr_tree_rev, &column_refs),
-            0.65
+            cost_model.get_filter_selectivity(expr_tree_shift1, &column_refs),
+            0.72
+        );
+        assert_approx_eq::assert_approx_eq!(
+            cost_model.get_filter_selectivity(expr_tree_shift2, &column_refs),
+            0.72
         );
     }