datafusion_physical_optimizer/
sanity_checker.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//! The [SanityCheckPlan] rule ensures that a given plan can
19//! accommodate its infinite sources, if there are any. It will reject
20//! non-runnable query plans that use pipeline-breaking operators on
21//! infinite input(s). In addition, it will check if all order and
22//! distribution requirements of a plan are satisfied by its children.
23
24use std::sync::Arc;
25
26use datafusion_common::Result;
27use datafusion_physical_plan::ExecutionPlan;
28
29use datafusion_common::config::{ConfigOptions, OptimizerOptions};
30use datafusion_common::plan_err;
31use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
32use datafusion_physical_expr::intervals::utils::{check_support, is_datatype_supported};
33use datafusion_physical_plan::execution_plan::{
34    Boundedness, EmissionType, InvariantLevel,
35};
36use datafusion_physical_plan::joins::SymmetricHashJoinExec;
37use datafusion_physical_plan::{
38    ChildSatisfactionOptions, ExecutionPlanProperties, get_plan_string,
39};
40
41use crate::PhysicalOptimizerRule;
42use datafusion_physical_expr_common::sort_expr::format_physical_sort_requirement_list;
43use itertools::izip;
44
45/// The SanityCheckPlan rule rejects the following query plans:
46/// 1. Invalid plans containing nodes whose order and/or distribution requirements
47///    are not satisfied by their children.
48/// 2. Plans that use pipeline-breaking operators on infinite input(s),
49///    it is impossible to execute such queries (they will never generate output nor finish)
50#[derive(Default, Debug)]
51pub struct SanityCheckPlan {}
52
53impl SanityCheckPlan {
54    #[expect(missing_docs)]
55    pub fn new() -> Self {
56        Self {}
57    }
58}
59
60impl PhysicalOptimizerRule for SanityCheckPlan {
61    fn optimize(
62        &self,
63        plan: Arc<dyn ExecutionPlan>,
64        config: &ConfigOptions,
65    ) -> Result<Arc<dyn ExecutionPlan>> {
66        check_plan_sanity_recursive(&plan, &config.optimizer)?;
67        Ok(plan)
68    }
69
70    fn name(&self) -> &str {
71        "SanityCheckPlan"
72    }
73
74    fn schema_check(&self) -> bool {
75        true
76    }
77}
78
79/// Bottom-up (post-order) read-only traversal that checks plan sanity.
80#[cfg_attr(feature = "recursive_protection", recursive::recursive)]
81fn check_plan_sanity_recursive(
82    plan: &Arc<dyn ExecutionPlan>,
83    optimizer_options: &OptimizerOptions,
84) -> Result<TreeNodeRecursion> {
85    plan.apply_children(|child| check_plan_sanity_recursive(child, optimizer_options))?;
86    check_plan_sanity(plan, optimizer_options)?;
87    Ok(TreeNodeRecursion::Continue)
88}
89
90/// This function propagates finiteness information and rejects any plan with
91/// pipeline-breaking operators acting on infinite inputs.
92pub fn check_finiteness_requirements(
93    input: &dyn ExecutionPlan,
94    optimizer_options: &OptimizerOptions,
95) -> Result<()> {
96    if let Some(exec) = input.downcast_ref::<SymmetricHashJoinExec>()
97        && !(optimizer_options.allow_symmetric_joins_without_pruning
98            || (exec.check_if_order_information_available()? && is_prunable(exec)))
99    {
100        return plan_err!(
101            "Join operation cannot operate on a non-prunable stream without enabling \
102                              the 'allow_symmetric_joins_without_pruning' configuration flag"
103        );
104    }
105
106    if matches!(
107        input.boundedness(),
108        Boundedness::Unbounded {
109            requires_infinite_memory: true
110        }
111    ) || (input.boundedness().is_unbounded()
112        && input.pipeline_behavior() == EmissionType::Final)
113    {
114        plan_err!(
115            "Cannot execute pipeline breaking queries, operator: {:?}",
116            input
117        )
118    } else {
119        Ok(())
120    }
121}
122
123/// This function returns whether a given symmetric hash join is amenable to
124/// data pruning. For this to be possible, it needs to have a filter where
125/// all involved [`PhysicalExpr`]s, [`Operator`]s and data types support
126/// interval calculations.
127///
128/// [`PhysicalExpr`]: datafusion_physical_plan::PhysicalExpr
129/// [`Operator`]: datafusion_expr::Operator
130fn is_prunable(join: &SymmetricHashJoinExec) -> bool {
131    join.filter().is_some_and(|filter| {
132        check_support(filter.expression(), &join.schema())
133            && filter
134                .schema()
135                .fields()
136                .iter()
137                .all(|f| is_datatype_supported(f.data_type()))
138    })
139}
140
141/// Ensures that the plan is pipeline friendly and the order and
142/// distribution requirements from its children are satisfied.
143pub fn check_plan_sanity(
144    plan: &Arc<dyn ExecutionPlan>,
145    optimizer_options: &OptimizerOptions,
146) -> Result<()> {
147    check_finiteness_requirements(plan.as_ref(), optimizer_options)?;
148    let input_distributions = plan.input_distribution_requirements();
149
150    for ((idx, child), sort_req, dist_req) in izip!(
151        plan.children().into_iter().enumerate(),
152        plan.required_input_ordering(),
153        input_distributions.per_child_distributions(),
154    ) {
155        let child_eq_props = child.equivalence_properties();
156        if let Some(sort_req) = sort_req {
157            let sort_req = sort_req.into_single();
158            if !child_eq_props.ordering_satisfy_requirement(sort_req.clone())? {
159                let plan_str = get_plan_string(plan);
160                return plan_err!(
161                    "Plan: {:?} does not satisfy order requirements: {}. Child-{} order: {}",
162                    plan_str,
163                    format_physical_sort_requirement_list(&sort_req),
164                    idx,
165                    child_eq_props.oeq_class()
166                );
167            }
168        }
169
170        if !input_distributions
171            .child_satisfaction(
172                idx,
173                child.as_ref(),
174                ChildSatisfactionOptions::new().with_allow_subset(true),
175            )?
176            .is_satisfied()
177        {
178            let plan_str = get_plan_string(plan);
179            return plan_err!(
180                "Plan: {:?} does not satisfy distribution requirements: {}. Child-{} output partitioning: {}",
181                plan_str,
182                dist_req,
183                idx,
184                child.output_partitioning()
185            );
186        }
187    }
188
189    plan.check_invariants(InvariantLevel::Executable)?;
190
191    Ok(())
192}
193
194// See tests in datafusion/core/tests/physical_optimizer
datafusion_physical_optimizer/sanity_checker.rs

datafusion_physical_optimizer/
sanity_checker.rs
