datafusion_optimizer/analyzer/
mod.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! [`Analyzer`] and [`AnalyzerRule`]
19
20use std::fmt::Debug;
21use std::sync::Arc;
22
23use log::debug;
24
25use datafusion_common::config::ConfigOptions;
26use datafusion_common::instant::Instant;
27use datafusion_common::Result;
28use datafusion_expr::expr_rewriter::FunctionRewrite;
29use datafusion_expr::{InvariantLevel, LogicalPlan};
30
31use crate::analyzer::resolve_grouping_function::ResolveGroupingFunction;
32use crate::analyzer::type_coercion::TypeCoercion;
33use crate::utils::log_plan;
34
35use self::function_rewrite::ApplyFunctionRewrites;
36
37pub mod function_rewrite;
38pub mod resolve_grouping_function;
39pub mod type_coercion;
40
41/// [`AnalyzerRule`]s transform [`LogicalPlan`]s in some way to make
42/// the plan valid prior to the rest of the DataFusion optimization process.
43///
44/// `AnalyzerRule`s are different than an [`OptimizerRule`](crate::OptimizerRule)s
45/// which must preserve the semantics of the `LogicalPlan`, while computing
46/// results in a more optimal way.
47///
48/// For example, an `AnalyzerRule` may resolve [`Expr`](datafusion_expr::Expr)s into more specific
49/// forms such as a subquery reference, or do type coercion to ensure the types
50/// of operands are correct.
51///
52/// Use [`SessionState::add_analyzer_rule`] to register additional
53/// `AnalyzerRule`s.
54///
55/// [`SessionState::add_analyzer_rule`]: https://docs.rs/datafusion/latest/datafusion/execution/session_state/struct.SessionState.html#method.add_analyzer_rule
56pub trait AnalyzerRule: Debug {
57    /// Rewrite `plan`
58    fn analyze(&self, plan: LogicalPlan, config: &ConfigOptions) -> Result<LogicalPlan>;
59
60    /// A human readable name for this analyzer rule
61    fn name(&self) -> &str;
62}
63
64/// Rule-based Analyzer.
65///
66/// Applies [`FunctionRewrite`]s and [`AnalyzerRule`]s to transform a
67/// [`LogicalPlan`] in preparation for execution.
68///
69/// For example, the `Analyzer` applies type coercion to ensure the types of
70/// operands match the types required by functions.
71#[derive(Clone, Debug)]
72pub struct Analyzer {
73    /// Expr --> Function writes to apply prior to analysis passes
74    pub function_rewrites: Vec<Arc<dyn FunctionRewrite + Send + Sync>>,
75    /// All rules to apply
76    pub rules: Vec<Arc<dyn AnalyzerRule + Send + Sync>>,
77}
78
79impl Default for Analyzer {
80    fn default() -> Self {
81        Self::new()
82    }
83}
84
85impl Analyzer {
86    /// Create a new analyzer using the recommended list of rules
87    pub fn new() -> Self {
88        let rules: Vec<Arc<dyn AnalyzerRule + Send + Sync>> = vec![
89            Arc::new(ResolveGroupingFunction::new()),
90            Arc::new(TypeCoercion::new()),
91        ];
92        Self::with_rules(rules)
93    }
94
95    /// Create a new analyzer with the given rules
96    pub fn with_rules(rules: Vec<Arc<dyn AnalyzerRule + Send + Sync>>) -> Self {
97        Self {
98            function_rewrites: vec![],
99            rules,
100        }
101    }
102
103    /// Add a function rewrite rule
104    pub fn add_function_rewrite(
105        &mut self,
106        rewrite: Arc<dyn FunctionRewrite + Send + Sync>,
107    ) {
108        self.function_rewrites.push(rewrite);
109    }
110
111    /// return the list of function rewrites in this analyzer
112    pub fn function_rewrites(&self) -> &[Arc<dyn FunctionRewrite + Send + Sync>] {
113        &self.function_rewrites
114    }
115
116    /// Analyze the logical plan by applying analyzer rules, and
117    /// do necessary check and fail the invalid plans
118    pub fn execute_and_check<F>(
119        &self,
120        plan: LogicalPlan,
121        config: &ConfigOptions,
122        mut observer: F,
123    ) -> Result<LogicalPlan>
124    where
125        F: FnMut(&LogicalPlan, &dyn AnalyzerRule),
126    {
127        // verify the logical plan required invariants at the start, before analyzer
128        plan.check_invariants(InvariantLevel::Always)
129            .map_err(|e| e.context("Invalid input plan passed to Analyzer"))?;
130
131        let start_time = Instant::now();
132        let mut new_plan = plan;
133
134        // Create an analyzer pass that rewrites `Expr`s to function_calls, as
135        // appropriate.
136        //
137        // Note this is run before all other rules since it rewrites based on
138        // the argument types (List or Scalar), and TypeCoercion may cast the
139        // argument types from Scalar to List.
140        let expr_to_function: Option<Arc<dyn AnalyzerRule + Send + Sync>> =
141            if self.function_rewrites.is_empty() {
142                None
143            } else {
144                Some(Arc::new(ApplyFunctionRewrites::new(
145                    self.function_rewrites.clone(),
146                )))
147            };
148        let rules = expr_to_function.iter().chain(self.rules.iter());
149
150        // TODO add common rule executor for Analyzer and Optimizer
151        for rule in rules {
152            new_plan = rule
153                .analyze(new_plan, config)
154                .map_err(|e| e.context(rule.name()))?;
155            log_plan(rule.name(), &new_plan);
156            observer(&new_plan, rule.as_ref());
157        }
158
159        // verify at the end, after the last LP analyzer pass, that the plan is executable.
160        new_plan
161            .check_invariants(InvariantLevel::Executable)
162            .map_err(|e| e.context("Invalid (non-executable) plan after Analyzer"))?;
163
164        log_plan("Final analyzed plan", &new_plan);
165        debug!("Analyzer took {} ms", start_time.elapsed().as_millis());
166        Ok(new_plan)
167    }
168}
datafusion_optimizer/analyzer/mod.rs

datafusion_optimizer/analyzer/
mod.rs
