datafusion_optimizer/
utils.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//! Utility functions leveraged by the query optimizer rules
19
20use std::collections::{BTreeSet, HashMap, HashSet};
21
22use crate::analyzer::type_coercion::TypeCoercionRewriter;
23use arrow::array::{new_null_array, Array, RecordBatch};
24use arrow::datatypes::{DataType, Field, Schema};
25use datafusion_common::cast::as_boolean_array;
26use datafusion_common::tree_node::{TransformedResult, TreeNode};
27use datafusion_common::{Column, DFSchema, Result, ScalarValue};
28use datafusion_expr::execution_props::ExecutionProps;
29use datafusion_expr::expr_rewriter::replace_col;
30use datafusion_expr::{logical_plan::LogicalPlan, ColumnarValue, Expr};
31use datafusion_physical_expr::create_physical_expr;
32use log::{debug, trace};
33use std::sync::Arc;
34
35/// Re-export of `NamesPreserver` for backwards compatibility,
36/// as it was initially placed here and then moved elsewhere.
37pub use datafusion_expr::expr_rewriter::NamePreserver;
38
39/// Returns true if `expr` contains all columns in `schema_cols`
40pub(crate) fn has_all_column_refs(expr: &Expr, schema_cols: &HashSet<Column>) -> bool {
41    let column_refs = expr.column_refs();
42    // note can't use HashSet::intersect because of different types (owned vs References)
43    schema_cols
44        .iter()
45        .filter(|c| column_refs.contains(c))
46        .count()
47        == column_refs.len()
48}
49
50pub(crate) fn replace_qualified_name(
51    expr: Expr,
52    cols: &BTreeSet<Column>,
53    subquery_alias: &str,
54) -> Result<Expr> {
55    let alias_cols: Vec<Column> = cols
56        .iter()
57        .map(|col| Column::new(Some(subquery_alias), &col.name))
58        .collect();
59    let replace_map: HashMap<&Column, &Column> =
60        cols.iter().zip(alias_cols.iter()).collect();
61
62    replace_col(expr, &replace_map)
63}
64
65/// Log the plan in debug/tracing mode after some part of the optimizer runs
66pub fn log_plan(description: &str, plan: &LogicalPlan) {
67    debug!("{description}:\n{}\n", plan.display_indent());
68    trace!("{description}::\n{}\n", plan.display_indent_schema());
69}
70
71/// Determine whether a predicate can restrict NULLs. e.g.
72/// `c0 > 8` return true;
73/// `c0 IS NULL` return false.
74pub fn is_restrict_null_predicate<'a>(
75    predicate: Expr,
76    join_cols_of_predicate: impl IntoIterator<Item = &'a Column>,
77) -> Result<bool> {
78    if matches!(predicate, Expr::Column(_)) {
79        return Ok(true);
80    }
81
82    // If result is single `true`, return false;
83    // If result is single `NULL` or `false`, return true;
84    Ok(
85        match evaluate_expr_with_null_column(predicate, join_cols_of_predicate)? {
86            ColumnarValue::Array(array) => {
87                if array.len() == 1 {
88                    let boolean_array = as_boolean_array(&array)?;
89                    boolean_array.is_null(0) || !boolean_array.value(0)
90                } else {
91                    false
92                }
93            }
94            ColumnarValue::Scalar(scalar) => matches!(
95                scalar,
96                ScalarValue::Boolean(None) | ScalarValue::Boolean(Some(false))
97            ),
98        },
99    )
100}
101
102/// Determines if an expression will always evaluate to null.
103/// `c0 + 8` return true
104/// `c0 IS NULL` return false
105/// `CASE WHEN c0 > 1 then 0 else 1` return false
106pub fn evaluates_to_null<'a>(
107    predicate: Expr,
108    null_columns: impl IntoIterator<Item = &'a Column>,
109) -> Result<bool> {
110    if matches!(predicate, Expr::Column(_)) {
111        return Ok(true);
112    }
113
114    Ok(
115        match evaluate_expr_with_null_column(predicate, null_columns)? {
116            ColumnarValue::Array(_) => false,
117            ColumnarValue::Scalar(scalar) => scalar.is_null(),
118        },
119    )
120}
121
122fn evaluate_expr_with_null_column<'a>(
123    predicate: Expr,
124    null_columns: impl IntoIterator<Item = &'a Column>,
125) -> Result<ColumnarValue> {
126    static DUMMY_COL_NAME: &str = "?";
127    let schema = Schema::new(vec![Field::new(DUMMY_COL_NAME, DataType::Null, true)]);
128    let input_schema = DFSchema::try_from(schema.clone())?;
129    let column = new_null_array(&DataType::Null, 1);
130    let input_batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![column])?;
131    let execution_props = ExecutionProps::default();
132    let null_column = Column::from_name(DUMMY_COL_NAME);
133
134    let join_cols_to_replace = null_columns
135        .into_iter()
136        .map(|column| (column, &null_column))
137        .collect::<HashMap<_, _>>();
138
139    let replaced_predicate = replace_col(predicate, &join_cols_to_replace)?;
140    let coerced_predicate = coerce(replaced_predicate, &input_schema)?;
141    create_physical_expr(&coerced_predicate, &input_schema, &execution_props)?
142        .evaluate(&input_batch)
143}
144
145fn coerce(expr: Expr, schema: &DFSchema) -> Result<Expr> {
146    let mut expr_rewrite = TypeCoercionRewriter { schema };
147    expr.rewrite(&mut expr_rewrite).data()
148}
149
150#[cfg(test)]
151mod tests {
152    use super::*;
153    use datafusion_expr::{binary_expr, case, col, in_list, is_null, lit, Operator};
154
155    #[test]
156    fn expr_is_restrict_null_predicate() -> Result<()> {
157        let test_cases = vec![
158            // a
159            (col("a"), true),
160            // a IS NULL
161            (is_null(col("a")), false),
162            // a IS NOT NULL
163            (Expr::IsNotNull(Box::new(col("a"))), true),
164            // a = NULL
165            (
166                binary_expr(
167                    col("a"),
168                    Operator::Eq,
169                    Expr::Literal(ScalarValue::Null, None),
170                ),
171                true,
172            ),
173            // a > 8
174            (binary_expr(col("a"), Operator::Gt, lit(8i64)), true),
175            // a <= 8
176            (binary_expr(col("a"), Operator::LtEq, lit(8i32)), true),
177            // CASE a WHEN 1 THEN true WHEN 0 THEN false ELSE NULL END
178            (
179                case(col("a"))
180                    .when(lit(1i64), lit(true))
181                    .when(lit(0i64), lit(false))
182                    .otherwise(lit(ScalarValue::Null))?,
183                true,
184            ),
185            // CASE a WHEN 1 THEN true ELSE false END
186            (
187                case(col("a"))
188                    .when(lit(1i64), lit(true))
189                    .otherwise(lit(false))?,
190                true,
191            ),
192            // CASE a WHEN 0 THEN false ELSE true END
193            (
194                case(col("a"))
195                    .when(lit(0i64), lit(false))
196                    .otherwise(lit(true))?,
197                false,
198            ),
199            // (CASE a WHEN 0 THEN false ELSE true END) OR false
200            (
201                binary_expr(
202                    case(col("a"))
203                        .when(lit(0i64), lit(false))
204                        .otherwise(lit(true))?,
205                    Operator::Or,
206                    lit(false),
207                ),
208                false,
209            ),
210            // (CASE a WHEN 0 THEN true ELSE false END) OR false
211            (
212                binary_expr(
213                    case(col("a"))
214                        .when(lit(0i64), lit(true))
215                        .otherwise(lit(false))?,
216                    Operator::Or,
217                    lit(false),
218                ),
219                true,
220            ),
221            // a IN (1, 2, 3)
222            (
223                in_list(col("a"), vec![lit(1i64), lit(2i64), lit(3i64)], false),
224                true,
225            ),
226            // a NOT IN (1, 2, 3)
227            (
228                in_list(col("a"), vec![lit(1i64), lit(2i64), lit(3i64)], true),
229                true,
230            ),
231            // a IN (NULL)
232            (
233                in_list(
234                    col("a"),
235                    vec![Expr::Literal(ScalarValue::Null, None)],
236                    false,
237                ),
238                true,
239            ),
240            // a NOT IN (NULL)
241            (
242                in_list(col("a"), vec![Expr::Literal(ScalarValue::Null, None)], true),
243                true,
244            ),
245        ];
246
247        let column_a = Column::from_name("a");
248        for (predicate, expected) in test_cases {
249            let join_cols_of_predicate = std::iter::once(&column_a);
250            let actual =
251                is_restrict_null_predicate(predicate.clone(), join_cols_of_predicate)?;
252            assert_eq!(actual, expected, "{predicate}");
253        }
254
255        Ok(())
256    }
257}
datafusion_optimizer/utils.rs

datafusion_optimizer/
utils.rs
