refactor(optimizer): optimizers (#17697)

* refactor(optimizer): optimizers

* optimizers/distributed

* run -> optimize()

* optimizer/dynamic_sample/ -> optimizer/optimizers/hyper_dp/dynamic_sample/

* move cascade check from optimizer to cascade

* optimizer/rule -> optimizers/rule

Author: BohuTANG

src/query/service/tests/it/sql/planner/optimizer/agg_index_query_rewrite.rs

@@ -28,10 +28,10 @@ use databend_common_expression::TableField;
 use databend_common_expression::TableSchemaRefExt;
 use databend_common_meta_app::schema::CreateOption;
 use databend_common_sql::optimizer::ir::SExpr;
-use databend_common_sql::optimizer::optimizers::RecursiveOptimizer;
+use databend_common_sql::optimizer::optimizers::recursive::RecursiveOptimizer;
+use databend_common_sql::optimizer::optimizers::rule::RuleID;
+use databend_common_sql::optimizer::optimizers::rule::DEFAULT_REWRITE_RULES;
 use databend_common_sql::optimizer::OptimizerContext;
-use databend_common_sql::optimizer::RuleID;
-use databend_common_sql::optimizer::DEFAULT_REWRITE_RULES;
 use databend_common_sql::plans::AggIndexInfo;
 use databend_common_sql::plans::CreateTablePlan;
 use databend_common_sql::plans::Plan;
@@ -393,8 +393,8 @@ async fn test_query_rewrite_impl(format: &str) -> Result<()> {
         query.clear_applied_rules();
 
         let opt_ctx = OptimizerContext::new(ctx.clone(), metadata.clone());
-        let result =
-            RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::TryApplyAggIndex]).run(&query)?;
+        let result = RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::TryApplyAggIndex])
+            .optimize(&query)?;
         let agg_index = find_push_down_index_info(&result)?;
         assert_eq!(
             suite.is_matched,
@@ -449,7 +449,7 @@ async fn plan_sql(
     {
         let s_expr = if optimize {
             let opt_ctx = OptimizerContext::new(ctx.clone(), metadata.clone());
-            RecursiveOptimizer::new(opt_ctx.clone(), &DEFAULT_REWRITE_RULES).run(&s_expr)?
+            RecursiveOptimizer::new(opt_ctx.clone(), &DEFAULT_REWRITE_RULES).optimize(&s_expr)?
         } else {
             *s_expr
         };

src/query/sql/src/planner/binder/bind_mutation/mutation_expression.rs

@@ -36,7 +36,7 @@ use crate::binder::InternalColumnBinding;
 use crate::binder::MutationStrategy;
 use crate::binder::MutationType;
 use crate::optimizer::ir::SExpr;
-use crate::optimizer::operator::SubqueryRewriter;
+use crate::optimizer::optimizers::operator::SubqueryRewriter;
 use crate::optimizer::OptimizerContext;
 use crate::plans::BoundColumnRef;
 use crate::plans::Filter;
@@ -313,7 +313,7 @@ impl MutationExpression {
                     let opt_ctx =
                         OptimizerContext::new(binder.ctx.clone(), binder.metadata.clone());
                     let mut rewriter = SubqueryRewriter::new(opt_ctx, None);
-                    let s_expr = rewriter.rewrite(&s_expr)?;
+                    let s_expr = rewriter.optimize(&s_expr)?;
 
                     Ok(MutationExpressionBindResult {
                         input: s_expr,

src/query/sql/src/planner/binder/bind_table_reference/bind_join.rs

@@ -33,8 +33,8 @@ use crate::normalize_identifier;
 use crate::optimizer::ir::ColumnSet;
 use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
-use crate::optimizer::operator::FlattenInfo;
-use crate::optimizer::operator::SubqueryRewriter;
+use crate::optimizer::optimizers::operator::FlattenInfo;
+use crate::optimizer::optimizers::operator::SubqueryRewriter;
 use crate::optimizer::OptimizerContext;
 use crate::planner::binder::scalar::ScalarBinder;
 use crate::planner::binder::Binder;

src/query/sql/src/planner/optimizer/distributed/distributed.rs

@@ -20,9 +20,9 @@ use databend_common_exception::Result;
 use crate::optimizer::ir::expr::PatternExtractor;
 use crate::optimizer::ir::group::Group;
 use crate::optimizer::ir::Memo;
-use crate::optimizer::rule::AppliedRules;
-use crate::optimizer::rule::RulePtr;
-use crate::optimizer::rule::TransformResult;
+use crate::optimizer::optimizers::rule::AppliedRules;
+use crate::optimizer::optimizers::rule::RulePtr;
+use crate::optimizer::optimizers::rule::TransformResult;
 use crate::plans::RelOperator;
 use crate::IndexType;
 

src/query/sql/src/planner/optimizer/ir/expr/m_expr.rs

@@ -24,8 +24,8 @@ use educe::Educe;
 use crate::optimizer::ir::property::RelExpr;
 use crate::optimizer::ir::property::RelationalProperty;
 use crate::optimizer::ir::StatInfo;
-use crate::optimizer::rule::AppliedRules;
-use crate::optimizer::rule::RuleID;
+use crate::optimizer::optimizers::rule::AppliedRules;
+use crate::optimizer::optimizers::rule::RuleID;
 use crate::plans::Exchange;
 use crate::plans::Operator;
 use crate::plans::RelOperator;

src/query/sql/src/planner/optimizer/ir/expr/s_expr.rs

@@ -13,25 +13,16 @@
 // limitations under the License.
 
 mod cost;
-mod distributed;
 pub mod ir;
-pub mod operator;
 #[allow(clippy::module_inception)]
 mod optimizer;
 mod optimizer_context;
 pub mod optimizers;
-pub mod rule;
 mod statistics;
 mod util;
 
-mod dynamic_sample;
-
 pub use optimizer::optimize;
 pub use optimizer::optimize_query;
 pub use optimizer_context::OptimizerContext;
-pub use rule::agg_index;
-pub use rule::RuleFactory;
-pub use rule::RuleID;
-pub use rule::RuleSet;
-pub use rule::DEFAULT_REWRITE_RULES;
+pub use optimizers::rule::agg_index;
 pub use util::contains_local_table_scan;

src/query/sql/src/planner/optimizer/mod.rs

@@ -23,26 +23,24 @@ use log::info;
 use crate::binder::target_probe;
 use crate::binder::MutationStrategy;
 use crate::binder::MutationType;
-use crate::optimizer::distributed::optimize_distributed_query;
-use crate::optimizer::distributed::BroadcastToShuffleOptimizer;
-use crate::optimizer::distributed::SortAndLimitPushDownOptimizer;
 use crate::optimizer::ir::Memo;
 use crate::optimizer::ir::SExpr;
-use crate::optimizer::operator::DeduplicateJoinConditionOptimizer;
-use crate::optimizer::operator::PullUpFilterOptimizer;
-use crate::optimizer::operator::RuleNormalizeAggregateOptimizer;
-use crate::optimizer::operator::RuleStatsAggregateOptimizer;
-use crate::optimizer::operator::SingleToInnerOptimizer;
-use crate::optimizer::operator::SubqueryRewriter;
+use crate::optimizer::optimizers::distributed::BroadcastToShuffleOptimizer;
+use crate::optimizer::optimizers::operator::DeduplicateJoinConditionOptimizer;
+use crate::optimizer::optimizers::operator::PullUpFilterOptimizer;
+use crate::optimizer::optimizers::operator::RuleNormalizeAggregateOptimizer;
+use crate::optimizer::optimizers::operator::RuleStatsAggregateOptimizer;
+use crate::optimizer::optimizers::operator::SingleToInnerOptimizer;
+use crate::optimizer::optimizers::operator::SubqueryRewriter;
+use crate::optimizer::optimizers::recursive::RecursiveOptimizer;
+use crate::optimizer::optimizers::rule::RuleID;
+use crate::optimizer::optimizers::rule::DEFAULT_REWRITE_RULES;
 use crate::optimizer::optimizers::CascadesOptimizer;
 use crate::optimizer::optimizers::DPhpy;
-use crate::optimizer::optimizers::RecursiveOptimizer;
 use crate::optimizer::statistics::CollectStatisticsOptimizer;
 use crate::optimizer::util::contains_local_table_scan;
 use crate::optimizer::util::contains_warehouse_table_scan;
 use crate::optimizer::OptimizerContext;
-use crate::optimizer::RuleID;
-use crate::optimizer::DEFAULT_REWRITE_RULES;
 use crate::plans::ConstantTableScan;
 use crate::plans::CopyIntoLocationPlan;
 use crate::plans::Join;
@@ -97,7 +95,7 @@ pub async fn optimize(opt_ctx: Arc<OptimizerContext>, plan: Plan) -> Result<Plan
                 let mut s_expr = s_expr;
                 if s_expr.contain_subquery() {
                     s_expr =
-                        Box::new(SubqueryRewriter::new(opt_ctx.clone(), None).rewrite(&s_expr)?);
+                        Box::new(SubqueryRewriter::new(opt_ctx.clone(), None).optimize(&s_expr)?);
                 }
                 Ok(Plan::Explain {
                     kind,
@@ -260,77 +258,53 @@ pub async fn optimize_query(opt_ctx: Arc<OptimizerContext>, mut s_expr: SExpr) -
     }
 
     // 2. Eliminate subqueries by rewriting them into more efficient forms
-    s_expr = SubqueryRewriter::new(opt_ctx.clone(), None).rewrite(&s_expr)?;
+    s_expr = SubqueryRewriter::new(opt_ctx.clone(), None).optimize(&s_expr)?;
 
     // 3. Apply statistics aggregation to gather and propagate statistics
     s_expr = RuleStatsAggregateOptimizer::new(opt_ctx.clone())
-        .run(&s_expr)
+        .optimize(&s_expr)
         .await?;
 
     // 4. Collect statistics for SExpr nodes to support cost estimation
     s_expr = CollectStatisticsOptimizer::new(opt_ctx.clone())
-        .run(&s_expr)
+        .optimize(&s_expr)
         .await?;
 
     // 5. Normalize aggregate, it should be executed before RuleSplitAggregate.
-    s_expr = RuleNormalizeAggregateOptimizer::new().run(&s_expr)?;
+    s_expr = RuleNormalizeAggregateOptimizer::new().optimize(&s_expr)?;
 
     // 6. Pull up and infer filter.
-    s_expr = PullUpFilterOptimizer::new(opt_ctx.clone()).run(&s_expr)?;
+    s_expr = PullUpFilterOptimizer::new(opt_ctx.clone()).optimize(&s_expr)?;
 
     // 7. Run default rewrite rules
-    s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &DEFAULT_REWRITE_RULES).run(&s_expr)?;
+    s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &DEFAULT_REWRITE_RULES).optimize(&s_expr)?;
 
     // 8. Run post rewrite rules
-    s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::SplitAggregate]).run(&s_expr)?;
+    s_expr =
+        RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::SplitAggregate]).optimize(&s_expr)?;
 
     // 9. Apply DPhyp algorithm for cost-based join reordering
-    if opt_ctx.get_enable_dphyp() && opt_ctx.get_enable_join_reorder() {
-        s_expr = DPhpy::new(opt_ctx.clone()).optimize(&s_expr).await?;
-    }
+    s_expr = DPhpy::new(opt_ctx.clone()).optimize(&s_expr).await?;
 
     // 10. After join reorder, Convert some single join to inner join.
-    s_expr = SingleToInnerOptimizer::new().run(&s_expr)?;
+    s_expr = SingleToInnerOptimizer::new().optimize(&s_expr)?;
 
     // 11. Deduplicate join conditions.
-    s_expr = DeduplicateJoinConditionOptimizer::new().run(&s_expr)?;
+    s_expr = DeduplicateJoinConditionOptimizer::new().optimize(&s_expr)?;
 
     // 12. Apply join commutativity to further optimize join ordering
     if opt_ctx.get_enable_join_reorder() {
         s_expr = RecursiveOptimizer::new(opt_ctx.clone(), [RuleID::CommuteJoin].as_slice())
-            .run(&s_expr)?;
+            .optimize(&s_expr)?;
     }
 
     // 13. Cascades optimizer may fail due to timeout, fallback to heuristic optimizer in this case.
-    let mut cascades = CascadesOptimizer::new(opt_ctx.clone())?;
-    s_expr = match cascades.optimize(s_expr.clone()) {
-        Ok(mut s_expr) => {
-            // 14. Push down sort and limit operations for distributed execution
-            // TODO(leiysky): do this optimization in cascades optimizer
-            if opt_ctx.get_enable_distributed_optimization() {
-                let sort_and_limit_optimizer = SortAndLimitPushDownOptimizer::create();
-                s_expr = sort_and_limit_optimizer.optimize(&s_expr)?;
-            }
-            s_expr
-        }
-
-        Err(e) => {
-            info!(
-                "CascadesOptimizer failed, fallback to heuristic optimizer: {}",
-                e
-            );
-            if opt_ctx.get_enable_distributed_optimization() {
-                s_expr = optimize_distributed_query(opt_ctx.get_table_ctx().clone(), &s_expr)?;
-            }
-
-            s_expr
-        }
-    };
+    s_expr = CascadesOptimizer::new(opt_ctx.clone())?.optimize(s_expr)?;
 
-    // 16. Eliminate unnecessary scalar calculations to clean up the final plan
+    // 14. Eliminate unnecessary scalar calculations to clean up the final plan
     if !opt_ctx.get_planning_agg_index() {
         s_expr = RecursiveOptimizer::new(opt_ctx.clone(), [RuleID::EliminateEvalScalar].as_slice())
-            .run(&s_expr)?;
+            .optimize(&s_expr)?;
     }
 
     Ok(s_expr)
@@ -355,24 +329,25 @@ async fn get_optimized_memo(opt_ctx: Arc<OptimizerContext>, mut s_expr: SExpr) -
 
     // Decorrelate subqueries, after this step, there should be no subquery in the expression.
     if s_expr.contain_subquery() {
-        s_expr = SubqueryRewriter::new(opt_ctx.clone(), None).rewrite(&s_expr)?;
+        s_expr = SubqueryRewriter::new(opt_ctx.clone(), None).optimize(&s_expr)?;
     }
 
     s_expr = RuleStatsAggregateOptimizer::new(opt_ctx.clone())
-        .run(&s_expr)
+        .optimize(&s_expr)
         .await?;
 
     // Collect statistics for each leaf node in SExpr.
     s_expr = CollectStatisticsOptimizer::new(opt_ctx.clone())
-        .run(&s_expr)
+        .optimize(&s_expr)
         .await?;
 
     // Pull up and infer filter.
-    s_expr = PullUpFilterOptimizer::new(opt_ctx.clone()).run(&s_expr)?;
+    s_expr = PullUpFilterOptimizer::new(opt_ctx.clone()).optimize(&s_expr)?;
     // Run default rewrite rules
-    s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &DEFAULT_REWRITE_RULES).run(&s_expr)?;
+    s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &DEFAULT_REWRITE_RULES).optimize(&s_expr)?;
     // Run post rewrite rules
-    s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::SplitAggregate]).run(&s_expr)?;
+    s_expr =
+        RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::SplitAggregate]).optimize(&s_expr)?;
 
     // Cost based optimization
     if opt_ctx.get_enable_dphyp() && opt_ctx.get_enable_join_reorder() {
@@ -388,7 +363,7 @@ async fn optimize_mutation(opt_ctx: Arc<OptimizerContext>, s_expr: SExpr) -> Res
     // Optimize the input plan.
     let mut input_s_expr = optimize_query(opt_ctx.clone(), s_expr.child(0)?.clone()).await?;
     input_s_expr = RecursiveOptimizer::new(opt_ctx.clone(), &[RuleID::MergeFilterIntoMutation])
-        .run(&input_s_expr)?;
+        .optimize(&input_s_expr)?;
 
     // For distributed query optimization, we need to remove the Exchange operator at the top of the plan.
     if let &RelOperator::Exchange(_) = input_s_expr.plan() {