refactor: Consolidate pushdown filter logic by renaming module and removing obsolete join handling, while enhancing optimizer rule tests

Author: longshan.lu

qurious/src/execution/session.rs

@@ -383,7 +383,8 @@ mod tests {
         println!("++++++++++++++");
         let batch = session
             .sql(
-                "select
+                "
+select
     s_acctbal,
     s_name,
     n_name,
@@ -426,12 +427,11 @@ order by
     n_name,
     s_name,
     p_partkey
-limit 10;",
+limit 10;
+",
             )
             .unwrap();
 
-        // let batch = session.sql("select avg(l_quantity) as count_order from lineitem")?;
-
         print_batches(&batch)?;
 
         Ok(())

qurious/src/optimizer/rule/eliminate_cross_join.rs

@@ -37,8 +37,10 @@ impl OptimizerRule for EliminateCrossJoin {
         input.apply(|plan| {
             if let LogicalPlan::CrossJoin(cross_join) = plan {
                 all_corss_joins.push(cross_join);
+                return Ok(TreeNodeRecursion::Continue);
             }
-            Ok(TreeNodeRecursion::Continue)
+
+            Ok(TreeNodeRecursion::Stop)
         })?;
 
         if all_corss_joins.is_empty() {
@@ -297,7 +299,7 @@ mod tests {
     }
 
     #[test]
-    fn test_tpch_03() {
+    fn test_eliminate_nested_cross_join() {
         assert_after_optimizer(
             "    select
         l_orderkey,

qurious/src/optimizer/rule/mod.rs

@@ -1,7 +1,7 @@
 mod count_wildcard_rule;
 mod eliminate_cross_join;
 mod extract_equijoin_predicate;
-mod pushdown_filter_join;
+mod pushdown_filter;
 mod rule_optimizer;
 mod scalar_subquery_to_join;
 mod simplify_exprs;
@@ -11,6 +11,6 @@ pub use rule_optimizer::*;
 
 pub use count_wildcard_rule::*;
 pub use extract_equijoin_predicate::*;
-pub use pushdown_filter_join::*;
+pub use pushdown_filter::*;
 pub use scalar_subquery_to_join::*;
 pub use type_coercion::*;

…c/optimizer/rule/pushdown_filter_join.rs …us/src/optimizer/rule/pushdown_filter.rsqurious/src/optimizer/rule/pushdown_filter_join.rs renamed to qurious/src/optimizer/rule/pushdown_filter.rs qurious/src/optimizer/rule/pushdown_filter_join.rs renamed to qurious/src/optimizer/rule/pushdown_filter.rs

@@ -249,7 +249,7 @@ mod tests {
             LogicalPlanBuilder,
         },
         optimizer::rule::{
-            eliminate_cross_join::EliminateCrossJoin, pushdown_filter_join::PushdownFilter, ExtractEquijoinPredicate,
+            eliminate_cross_join::EliminateCrossJoin, pushdown_filter::PushdownFilter, ExtractEquijoinPredicate,
         },
         test_utils::{assert_after_optimizer, assert_after_optimizer_with_plan},
     };

qurious/src/optimizer/rule/rule_optimizer.rs

@@ -6,7 +6,7 @@ use crate::logical::plan::LogicalPlan;
 use crate::optimizer::rule::count_wildcard_rule::CountWildcardRule;
 use crate::optimizer::rule::eliminate_cross_join::EliminateCrossJoin;
 use crate::optimizer::rule::extract_equijoin_predicate::ExtractEquijoinPredicate;
-use crate::optimizer::rule::pushdown_filter_join::PushdownFilter;
+use crate::optimizer::rule::pushdown_filter::PushdownFilter;
 use crate::optimizer::rule::scalar_subquery_to_join::ScalarSubqueryToJoin;
 use crate::optimizer::rule::simplify_exprs::SimplifyExprs;
 use crate::optimizer::rule::type_coercion::TypeCoercion;

qurious/src/physical/plan/aggregate/hash.rs

@@ -7,7 +7,7 @@ use crate::physical::{
     plan::PhysicalPlan,
 };
 use crate::utils::array::create_hashes;
-use arrow::compute::TakeOptions;
+use arrow::compute::{concat_batches, TakeOptions};
 use arrow::row::{RowConverter, SortField};
 use arrow::{
     array::{ArrayRef, UInt64Array},
@@ -143,24 +143,25 @@ impl PhysicalPlan for HashAggregate {
                 .map(|e| e.create_accumulator())
                 .collect::<Result<Vec<_>>>()
         };
-        let mut group_accumulator = GroupAccumulator::try_new(&accumlator_factory)?;
-        // for each batch from the input executor
-        for batch in batches {
-            // evaluate the groupt expression
-            let group_by_values = self
-                .group_exprs
-                .iter()
-                .map(|e| e.evaluate(&batch))
-                .collect::<Result<Vec<ArrayRef>>>()?;
-            // evaluate the aggregate expression
-            let input_values = self
-                .aggregate_exprs
-                .iter()
-                .map(|e| e.expression().evaluate(&batch))
-                .collect::<Result<Vec<ArrayRef>>>()?;
-
-            group_accumulator.update(batch.num_rows(), &group_by_values, &input_values)?;
+        if batches.is_empty() {
+            return Ok(vec![]);
         }
+        let schema = batches[0].schema();
+        let batch = concat_batches(&schema, &batches)?;
+        let mut group_accumulator = GroupAccumulator::try_new(&accumlator_factory)?;
+        let group_by_values = self
+            .group_exprs
+            .iter()
+            .map(|e| e.evaluate(&batch))
+            .collect::<Result<Vec<ArrayRef>>>()?;
+        // evaluate the aggregate expression
+        let input_values = self
+            .aggregate_exprs
+            .iter()
+            .map(|e| e.expression().evaluate(&batch))
+            .collect::<Result<Vec<ArrayRef>>>()?;
+
+        group_accumulator.update(batch.num_rows(), &group_by_values, &input_values)?;
 
         let schema = self.schema.arrow_schema();
         RecordBatch::try_new(schema.clone(), group_accumulator.output(&schema)?)

qurious/tests/tpch/q2.slt

@@ -0,0 +1,50 @@
+query RTTITTTT
+select
+  s_acctbal,
+  s_name,
+  n_name,
+  p_partkey,
+  p_mfgr,
+  s_address,
+  s_phone,
+  s_comment
+from
+  part,
+  supplier,
+  partsupp,
+  nation,
+  region
+where
+  p_partkey = ps_partkey
+  and s_suppkey = ps_suppkey
+  and p_size = 15
+  and p_type like '%BRASS'
+  and s_nationkey = n_nationkey
+  and n_regionkey = r_regionkey
+  and r_name = 'EUROPE'
+  and ps_supplycost = (
+  select
+  min(ps_supplycost)
+  from
+  partsupp,
+  supplier,
+  nation,
+  region
+  where
+  p_partkey = ps_partkey
+  and s_suppkey = ps_suppkey
+  and s_nationkey = n_nationkey
+  and n_regionkey = r_regionkey
+  and r_name = 'EUROPE'
+)
+order by
+  s_acctbal desc,
+  n_name,
+  s_name,
+  p_partkey
+limit 10;
+----
+4186.95 Supplier#000000077 GERMANY 249 Manufacturer#4 wVtcr0uH3CyrSiWMLsqnB09Syo,UuZxPMeBghlY 17-281-345-4863 the slyly final asymptotes. blithely pending theodoli
+1883.37 Supplier#000000086 ROMANIA 1015 Manufacturer#4 J1fgg5QaqnN 29-903-665-7065 cajole furiously special, final requests: furiously spec
+1687.81 Supplier#000000017 ROMANIA 1634 Manufacturer#2 c2d,ESHRSkK3WYnxpgw6aOqN0q 29-601-884-9219 eep against the furiously bold ideas. fluffily bold packa
+287.16 Supplier#000000052 ROMANIA 323 Manufacturer#4 WCk XCHYzBA1dvJDSol4ZJQQcQN, 29-974-934-4713 dolites are slyly against the furiously regular packages. ironic, final deposits cajole quickly