refactor(optimizer): optimizer/operator (#17689)

BohuTANG · Copilot · web-flow · commit 8686791398f2 · 2025-04-02T12:34:03.000+08:00
* refactor(optimizer): operator

* optimizer/join -&gt; operator/join

* optimizer/decorrelate -&gt; operator/decorrelate

* decorrelate_subquery -&gt; SubqueryRewriter.rewrite

* optimizer/aggregate/ -&gt; operator/aggregate/

* opeartor/README.md

* Update src/query/sql/src/planner/optimizer/operator/decorrelate/subquery_rewriter.rs

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;

---------

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;
diff --git a/src/query/sql/src/planner/binder/bind_mutation/mutation_expression.rs b/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::SubqueryRewriter;
+use crate::optimizer::operator::SubqueryRewriter;
 use crate::plans::BoundColumnRef;
 use crate::plans::Filter;
 use crate::plans::Join;
diff --git a/src/query/sql/src/planner/binder/bind_table_reference/bind_join.rs b/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::FlattenInfo;
-use crate::optimizer::SubqueryRewriter;
+use crate::optimizer::operator::FlattenInfo;
+use crate::optimizer::operator::SubqueryRewriter;
 use crate::planner::binder::scalar::ScalarBinder;
 use crate::planner::binder::Binder;
 use crate::planner::semantic::NameResolutionContext;
diff --git a/src/query/sql/src/planner/optimizer/mod.rs b/src/query/sql/src/planner/optimizer/mod.rs
@@ -12,15 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod aggregate;
 mod cascades;
 mod cost;
-mod decorrelate;
 mod distributed;
-mod filter;
 mod hyper_dp;
 pub mod ir;
-mod join;
+pub mod operator;
 #[allow(clippy::module_inception)]
 mod optimizer;
 pub mod rule;
@@ -30,11 +27,6 @@ mod util;
 mod dynamic_sample;
 
 pub use cascades::CascadesOptimizer;
-pub use decorrelate::FlattenInfo;
-pub use decorrelate::SubqueryRewriter;
-pub use filter::DeduplicateJoinConditionOptimizer;
-pub use filter::InferFilterOptimizer;
-pub use filter::NormalizeDisjunctiveFilterOptimizer;
 pub use hyper_dp::DPhpy;
 pub use optimizer::optimize;
 pub use optimizer::optimize_query;
diff --git a/src/query/sql/src/planner/optimizer/operator/README.md b/src/query/sql/src/planner/optimizer/operator/README.md
@@ -0,0 +1,27 @@
+# Operator-Specific Optimizations
+
+This directory contains operator-specific optimization rules for the Databend query optimizer. Unlike the rule-based transformations in the `transforms/rule/` directory that focus on general transformation patterns, the optimizations here are specialized for specific logical operators.
+
+## Directory Structure
+
+The `operator/` directory is organized by operator type:
+
+- **aggregate/**: Optimizations specific to aggregate operations
+  - `normalize_aggregate.rs`: Normalizes aggregate expressions
+  - `stats_aggregate.rs`: Optimizes aggregates based on statistics
+
+- **decorrelate/**: Optimizations for handling correlated subqueries
+  - Contains utilities for subquery rewriting and unnesting
+
+- **filter/**: Optimizations for filter operations
+  - Includes deduplication of join conditions
+  - Inference of filter properties
+  - Normalization of disjunctive filters
+  - Pull-up filter optimizations
+
+- **join/**: Optimizations specific to join operations
+  - Includes transformations like converting single joins to inner joins
+
+## Usage
+
+These operator-specific optimizers are typically used in the query optimization pipeline to transform logical plans into more efficient forms. They focus on the semantics and properties of specific operators rather than general transformation patterns.
diff --git a/src/query/sql/src/planner/optimizer/operator/aggregate/mod.rs b/src/query/sql/src/planner/optimizer/operator/aggregate/mod.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/aggregate/normalize_aggregate.rs b/src/query/sql/src/planner/optimizer/operator/aggregate/normalize_aggregate.rs
@@ -171,3 +171,9 @@ impl RuleNormalizeAggregateOptimizer {
         }
     }
 }
+
+impl Default for RuleNormalizeAggregateOptimizer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
diff --git a/src/query/sql/src/planner/optimizer/operator/aggregate/stats_aggregate.rs b/src/query/sql/src/planner/optimizer/operator/aggregate/stats_aggregate.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/decorrelate/decorrelate.rs b/src/query/sql/src/planner/optimizer/operator/decorrelate/decorrelate.rs
@@ -17,7 +17,6 @@ use std::collections::HashSet;
 use std::sync::Arc;
 
 use databend_common_ast::Span;
-use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::type_check::common_super_type;
@@ -31,13 +30,13 @@ use databend_common_functions::BUILTIN_FUNCTIONS;
 use crate::binder::ColumnBindingBuilder;
 use crate::binder::JoinPredicate;
 use crate::binder::Visibility;
-use crate::optimizer::decorrelate::subquery_rewriter::FlattenInfo;
-use crate::optimizer::decorrelate::subquery_rewriter::SubqueryRewriter;
-use crate::optimizer::decorrelate::subquery_rewriter::UnnestResult;
 use crate::optimizer::ir::ColumnSet;
 use crate::optimizer::ir::Matcher;
 use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
+use crate::optimizer::operator::FlattenInfo;
+use crate::optimizer::operator::SubqueryRewriter;
+use crate::optimizer::operator::UnnestResult;
 use crate::plans::BoundColumnRef;
 use crate::plans::CastExpr;
 use crate::plans::ComparisonOp;
@@ -53,27 +52,6 @@ use crate::plans::ScalarExpr;
 use crate::plans::SubqueryExpr;
 use crate::plans::SubqueryType;
 use crate::IndexType;
-use crate::MetadataRef;
-
-/// Decorrelate subqueries inside `s_expr`.
-///
-/// We only need to process three kinds of join: Scalar Subquery, Any Subquery, and Exists Subquery.
-/// Other kinds of subqueries have be converted to one of the above subqueries in `type_check`.
-///
-/// It will rewrite `s_expr` to all kinds of join.
-/// Correlated scalar subquery -> Single join
-/// Any subquery -> Marker join
-/// Correlated exists subquery -> Marker join
-///
-/// More information can be found in the paper: Unnesting Arbitrary Queries
-pub fn decorrelate_subquery(
-    ctx: Arc<dyn TableContext>,
-    metadata: MetadataRef,
-    s_expr: SExpr,
-) -> Result<SExpr> {
-    let mut rewriter = SubqueryRewriter::new(ctx, metadata, None);
-    rewriter.rewrite(&s_expr)
-}
 
 impl SubqueryRewriter {
     // Try to decorrelate a `CrossApply` into `SemiJoin` or `AntiJoin`.
diff --git a/src/query/sql/src/planner/optimizer/operator/decorrelate/flatten_plan.rs b/src/query/sql/src/planner/optimizer/operator/decorrelate/flatten_plan.rs
@@ -25,11 +25,11 @@ use databend_common_expression::DataSchema;
 
 use crate::binder::ColumnBindingBuilder;
 use crate::binder::Visibility;
-use crate::optimizer::decorrelate::subquery_rewriter::FlattenInfo;
 use crate::optimizer::ir::ColumnSet;
 use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
-use crate::optimizer::SubqueryRewriter;
+use crate::optimizer::operator::FlattenInfo;
+use crate::optimizer::operator::SubqueryRewriter;
 use crate::plans::Aggregate;
 use crate::plans::AggregateFunction;
 use crate::plans::AggregateMode;
diff --git a/src/query/sql/src/planner/optimizer/operator/decorrelate/flatten_scalar.rs b/src/query/sql/src/planner/optimizer/operator/decorrelate/flatten_scalar.rs
@@ -17,7 +17,7 @@ use databend_common_exception::Result;
 
 use crate::binder::ColumnBindingBuilder;
 use crate::optimizer::ir::ColumnSet;
-use crate::optimizer::SubqueryRewriter;
+use crate::optimizer::operator::SubqueryRewriter;
 use crate::plans::AggregateFunction;
 use crate::plans::AggregateFunctionScalarSortDesc;
 use crate::plans::BoundColumnRef;
diff --git a/src/query/sql/src/planner/optimizer/operator/decorrelate/mod.rs b/src/query/sql/src/planner/optimizer/operator/decorrelate/mod.rs
@@ -18,6 +18,6 @@ mod flatten_plan;
 mod flatten_scalar;
 mod subquery_rewriter;
 
-pub use decorrelate::decorrelate_subquery;
 pub use subquery_rewriter::FlattenInfo;
 pub use subquery_rewriter::SubqueryRewriter;
+pub use subquery_rewriter::UnnestResult;
diff --git a/src/query/sql/src/planner/optimizer/operator/decorrelate/subquery_rewriter.rs b/src/query/sql/src/planner/optimizer/operator/decorrelate/subquery_rewriter.rs
@@ -86,6 +86,17 @@ impl SubqueryRewriter {
         }
     }
 
+    /// Decorrelate subqueries inside `s_expr`.
+    ///
+    /// We only need to process three kinds of join: Scalar Subquery, Any Subquery, and Exists Subquery.
+    /// Other kinds of subqueries have been converted to one of the above subqueries in `type_check`.
+    ///
+    /// It will rewrite `s_expr` to all kinds of join.
+    /// Correlated scalar subquery -> Single join
+    /// Any subquery -> Marker join
+    /// Correlated exists subquery -> Marker join
+    ///
+    /// More information can be found in the paper: Unnesting Arbitrary Queries
     #[recursive::recursive]
     pub fn rewrite(&mut self, s_expr: &SExpr) -> Result<SExpr> {
         match s_expr.plan().clone() {
diff --git a/src/query/sql/src/planner/optimizer/operator/filter/deduplicate_join_condition.rs b/src/query/sql/src/planner/optimizer/operator/filter/deduplicate_join_condition.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/filter/infer_filter.rs b/src/query/sql/src/planner/optimizer/operator/filter/infer_filter.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/filter/mod.rs b/src/query/sql/src/planner/optimizer/operator/filter/mod.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/filter/normalize_disjunctive_filter.rs b/src/query/sql/src/planner/optimizer/operator/filter/normalize_disjunctive_filter.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/filter/pull_up_filter.rs b/src/query/sql/src/planner/optimizer/operator/filter/pull_up_filter.rs
@@ -17,9 +17,9 @@ use std::sync::Arc;
 use databend_common_exception::Result;
 
 use crate::binder::split_conjunctions;
-use crate::optimizer::filter::InferFilterOptimizer;
-use crate::optimizer::filter::NormalizeDisjunctiveFilterOptimizer;
 use crate::optimizer::ir::SExpr;
+use crate::optimizer::operator::InferFilterOptimizer;
+use crate::optimizer::operator::NormalizeDisjunctiveFilterOptimizer;
 use crate::plans::EvalScalar;
 use crate::plans::Filter;
 use crate::plans::FunctionCall;
diff --git a/src/query/sql/src/planner/optimizer/operator/join/mod.rs b/src/query/sql/src/planner/optimizer/operator/join/mod.rs
diff --git a/src/query/sql/src/planner/optimizer/operator/join/single_to_inner.rs b/src/query/sql/src/planner/optimizer/operator/join/single_to_inner.rs
@@ -59,3 +59,9 @@ impl SingleToInnerOptimizer {
         Ok(s_expr)
     }
 }
+
+impl Default for SingleToInnerOptimizer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
diff --git a/src/query/sql/src/planner/optimizer/operator/mod.rs b/src/query/sql/src/planner/optimizer/operator/mod.rs
@@ -0,0 +1,30 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod aggregate;
+mod decorrelate;
+mod filter;
+mod join;
+
+pub use aggregate::RuleNormalizeAggregateOptimizer;
+pub use aggregate::RuleStatsAggregateOptimizer;
+pub use decorrelate::FlattenInfo;
+pub use decorrelate::SubqueryRewriter;
+pub use decorrelate::UnnestResult;
+pub use filter::DeduplicateJoinConditionOptimizer;
+pub use filter::InferFilterOptimizer;
+pub use filter::JoinProperty;
+pub use filter::NormalizeDisjunctiveFilterOptimizer;
+pub use filter::PullUpFilterOptimizer;
+pub use join::SingleToInnerOptimizer;
diff --git a/src/query/sql/src/planner/optimizer/optimizer.rs b/src/query/sql/src/planner/optimizer/optimizer.rs
@@ -22,22 +22,22 @@ use databend_common_exception::Result;
 use educe::Educe;
 use log::info;
 
-use super::aggregate::RuleStatsAggregateOptimizer;
-use super::distributed::BroadcastToShuffleOptimizer;
 use crate::binder::target_probe;
 use crate::binder::MutationStrategy;
 use crate::binder::MutationType;
-use crate::optimizer::aggregate::RuleNormalizeAggregateOptimizer;
 use crate::optimizer::cascades::CascadesOptimizer;
-use crate::optimizer::decorrelate::decorrelate_subquery;
 use crate::optimizer::distributed::optimize_distributed_query;
+use crate::optimizer::distributed::BroadcastToShuffleOptimizer;
 use crate::optimizer::distributed::SortAndLimitPushDownOptimizer;
-use crate::optimizer::filter::DeduplicateJoinConditionOptimizer;
-use crate::optimizer::filter::PullUpFilterOptimizer;
 use crate::optimizer::hyper_dp::DPhpy;
 use crate::optimizer::ir::Memo;
 use crate::optimizer::ir::SExpr;
-use crate::optimizer::join::SingleToInnerOptimizer;
+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::rule::TransformResult;
 use crate::optimizer::statistics::CollectStatisticsOptimizer;
 use crate::optimizer::util::contains_local_table_scan;
@@ -229,11 +229,14 @@ pub async fn optimize(mut opt_ctx: OptimizerContext, plan: Plan) -> Result<Plan>
 
                 let mut s_expr = s_expr;
                 if s_expr.contain_subquery() {
-                    s_expr = Box::new(decorrelate_subquery(
-                        opt_ctx.table_ctx.clone(),
-                        opt_ctx.metadata.clone(),
-                        *s_expr.clone(),
-                    )?);
+                    s_expr = Box::new(
+                        SubqueryRewriter::new(
+                            opt_ctx.table_ctx.clone(),
+                            opt_ctx.metadata.clone(),
+                            None,
+                        )
+                        .rewrite(&s_expr)?,
+                    );
                 }
                 Ok(Plan::Explain {
                     kind,
@@ -394,11 +397,8 @@ pub async fn optimize_query(opt_ctx: &mut 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 = decorrelate_subquery(
-            opt_ctx.table_ctx.clone(),
-            opt_ctx.metadata.clone(),
-            s_expr.clone(),
-        )?;
+        s_expr = SubqueryRewriter::new(opt_ctx.table_ctx.clone(), opt_ctx.metadata.clone(), None)
+            .rewrite(&s_expr)?;
     }
 
     s_expr = RuleStatsAggregateOptimizer::new(opt_ctx.table_ctx.clone(), opt_ctx.metadata.clone())
@@ -492,11 +492,8 @@ async fn get_optimized_memo(opt_ctx: &mut 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 = decorrelate_subquery(
-            opt_ctx.table_ctx.clone(),
-            opt_ctx.metadata.clone(),
-            s_expr.clone(),
-        )?;
+        s_expr = SubqueryRewriter::new(opt_ctx.table_ctx.clone(), opt_ctx.metadata.clone(), None)
+            .rewrite(&s_expr)?;
     }
 
     s_expr = RuleStatsAggregateOptimizer::new(opt_ctx.table_ctx.clone(), opt_ctx.metadata.clone())
diff --git a/src/query/sql/src/planner/optimizer/rule/rewrite/rule_push_down_filter_join.rs b/src/query/sql/src/planner/optimizer/rule/rewrite/rule_push_down_filter_join.rs
@@ -17,11 +17,11 @@ use std::sync::Arc;
 use databend_common_exception::Result;
 
 use crate::binder::JoinPredicate;
-use crate::optimizer::filter::InferFilterOptimizer;
-use crate::optimizer::filter::JoinProperty;
 use crate::optimizer::ir::Matcher;
 use crate::optimizer::ir::RelExpr;
 use crate::optimizer::ir::SExpr;
+use crate::optimizer::operator::InferFilterOptimizer;
+use crate::optimizer::operator::JoinProperty;
 use crate::optimizer::rule::constant::false_constant;
 use crate::optimizer::rule::constant::is_falsy;
 use crate::optimizer::rule::rewrite::push_down_filter_join::can_filter_null;
diff --git a/src/query/sql/tests/optimizer/filter/deduplicate_join_condition_test.rs b/src/query/sql/tests/optimizer/filter/deduplicate_join_condition_test.rs
@@ -19,7 +19,7 @@ use databend_common_exception::Result;
 use databend_common_expression::types::DataType;
 use databend_common_expression::types::NumberDataType;
 use databend_common_sql::optimizer::ir::SExpr;
-use databend_common_sql::optimizer::DeduplicateJoinConditionOptimizer;
+use databend_common_sql::optimizer::operator::DeduplicateJoinConditionOptimizer;
 use databend_common_sql::planner::binder::ColumnBinding;
 use databend_common_sql::planner::binder::Visibility;
 use databend_common_sql::planner::plans::BoundColumnRef;
diff --git a/src/query/sql/tests/optimizer/filter/infer_filter_test.rs b/src/query/sql/tests/optimizer/filter/infer_filter_test.rs
@@ -18,7 +18,7 @@ use databend_common_expression::types::DataType;
 use databend_common_expression::types::NumberDataType;
 use databend_common_expression::types::NumberScalar;
 use databend_common_expression::types::Scalar;
-use databend_common_sql::optimizer::InferFilterOptimizer;
+use databend_common_sql::optimizer::operator::InferFilterOptimizer;
 use databend_common_sql::planner::binder::ColumnBinding;
 use databend_common_sql::planner::binder::Visibility;
 use databend_common_sql::planner::plans::BoundColumnRef;
diff --git a/src/query/sql/tests/optimizer/filter/normalize_disjunctive_filter_test.rs b/src/query/sql/tests/optimizer/filter/normalize_disjunctive_filter_test.rs
@@ -17,7 +17,7 @@ use databend_common_expression::types::DataType;
 use databend_common_expression::types::NumberDataType;
 use databend_common_expression::types::NumberScalar;
 use databend_common_expression::Scalar;
-use databend_common_sql::optimizer::NormalizeDisjunctiveFilterOptimizer;
+use databend_common_sql::optimizer::operator::NormalizeDisjunctiveFilterOptimizer;
 use databend_common_sql::planner::binder::ColumnBinding;
 use databend_common_sql::planner::binder::Visibility;
 use databend_common_sql::planner::plans::BoundColumnRef;

Original file line number	Diff line number	Diff line change
`@@ -171,3 +171,9 @@ impl RuleNormalizeAggregateOptimizer {`
`171`	`171`	`}`
`172`	`172`	`}`
`173`	`173`	`}`
	`174`	`+`
	`175`	`+impl Default for RuleNormalizeAggregateOptimizer {`
	`176`	`+ fn default() -> Self {`
	`177`	`+ Self::new()`
	`178`	`+ }`
	`179`	`+}`