// SPDX-License-Identifier: Apache-2.0

// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership

// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

//! [`PruningPredicate`] to apply filter [`Expr`] to prune "containers"

//! based on statistics (e.g. Parquet Row Groups)

//!

//! [`Expr`]: https://docs.rs/datafusion/latest/datafusion/logical_expr/enum.Expr.html

use std::collections::HashSet;

use std::sync::Arc;

use crate::in_list::{SetMembership, unwrap_scalar};

use crate::key_range_bitmap_expr::KeyRangeBitmapPruningExpr;

use crate::primitive_in_list::PrimitiveInListDomain;

use crate::string_in_list::{BinaryInListPruningExpr, StringInListPruningExpr};

use arrow::array::AsArray;

use arrow::{

array::{ArrayRef, BooleanArray, new_null_array},

datatypes::{DataType, Field, Schema, SchemaRef},

record_batch::{RecordBatch, RecordBatchOptions},

};

// pub use for backwards compatibility

pub use datafusion_common::pruning::PruningStatistics;

use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;

use datafusion_physical_plan::metrics::Count;

use log::{debug, trace};

use datafusion_common::error::Result;

use datafusion_common::tree_node::{TransformedResult, TreeNodeRecursion};

use datafusion_common::{_internal_datafusion_err, Column, assert_eq_or_internal_err};

use datafusion_common::{

ScalarValue, internal_datafusion_err, plan_datafusion_err, plan_err,

tree_node::{Transformed, TreeNode},

};

use datafusion_expr_common::casts::try_cast_literal_to_type;

use datafusion_expr_common::operator::Operator;

use datafusion_physical_expr::utils::{Guarantee, LiteralGuarantee};

use datafusion_physical_expr::{PhysicalExprRef, expressions as phys_expr};

use datafusion_physical_expr_common::physical_expr::snapshot_physical_expr_opt;

use datafusion_physical_plan::joins::HashTableLookupExpr;

use datafusion_physical_plan::{ColumnarValue, PhysicalExpr};

/// Used to prove that arbitrary predicates (boolean expression) can not

/// possibly evaluate to `true` given information about a column provided by

/// [`PruningStatistics`].

///

/// # Introduction

///

/// `PruningPredicate` analyzes filter expressions using statistics such as

/// min/max values and null counts, attempting to prove a "container" (e.g.

/// Parquet Row Group) can be skipped without reading the actual data,

/// potentially leading to significant performance improvements.

///

/// For example, `PruningPredicate`s are used to prune Parquet Row Groups based

/// on the min/max values found in the Parquet metadata. If the

/// `PruningPredicate` can prove that the filter can never evaluate to `true`

/// for any row in the Row Group, the entire Row Group is skipped during query

/// execution.

///

/// The `PruningPredicate` API is general, and can be used for pruning other

/// types of containers (e.g. files) based on statistics that may be known from

/// external catalogs (e.g. Delta Lake) or other sources. How this works is a

/// subtle topic. See the Background and Implementation section for details.

///

/// `PruningPredicate` supports:

///

/// 1. Arbitrary expressions (including user defined functions)

///

/// 2. Vectorized evaluation (provide more than one set of statistics at a time)

/// so it is suitable for pruning 1000s of containers.

///

/// 3. Any source of information that implements the [`PruningStatistics`] trait

/// (not just Parquet metadata).

///

/// # Example

///

/// See the [`pruning.rs` example in the `datafusion-examples`] for a complete

/// example of how to use `PruningPredicate` to prune files based on min/max

/// values.

///

/// [`pruning.rs` example in the `datafusion-examples`]: https://github.com/apache/datafusion/blob/main/datafusion-examples/examples/query_planning/pruning.rs

///

/// Given an expression like `x = 5` and statistics for 3 containers (Row

/// Groups, files, etc) `A`, `B`, and `C`:

///

/// ```text

/// A: {x_min = 0, x_max = 4}

/// B: {x_min = 2, x_max = 10}

/// C: {x_min = 5, x_max = 8}

/// ```

///

/// `PruningPredicate` will conclude that the rows in container `A` can never

/// be true (as the maximum value is only `4`), so it can be pruned:

///

/// ```text

/// A: false (no rows could possibly match x = 5)

/// B: true (rows might match x = 5)

/// C: true (rows might match x = 5)

/// ```

///

/// See [`PruningPredicateBuilder`] and [`PruningPredicate::prune`] for more information.

///

/// # Background

///

/// ## Boolean Tri-state logic

///

/// To understand the details of the rest of this documentation, it is important

/// to understand how the tri-state boolean logic in SQL works. As this is

/// somewhat esoteric, we review it here.

///

/// SQL has a notion of `NULL` that represents the value is `“unknown”` and this

/// uncertainty propagates through expressions. SQL `NULL` behaves very

/// differently than the `NULL` in most other languages where it is a special,

/// sentinel value (e.g. `0` in `C/C++`). While representing uncertainty with

/// `NULL` is powerful and elegant, SQL `NULL`s are often deeply confusing when

/// first encountered as they behave differently than most programmers may

/// expect.

///

/// In most other programming languages,

/// * `a == NULL` evaluates to `true` if `a` also had the value `NULL`

/// * `a == NULL` evaluates to `false` if `a` has any other value

///

/// However, in SQL `a = NULL` **always** evaluates to `NULL` (never `true` or

/// `false`):

///

/// Expression | Result

/// ------------- | ---------

/// `1 = NULL` | `NULL`

/// `NULL = NULL` | `NULL`

///

/// Also important is how `AND` and `OR` works with tri-state boolean logic as

/// (perhaps counterintuitively) the result is **not** always NULL. While

/// consistent with the notion of `NULL` representing “unknown”, this is again,

/// often deeply confusing 🤯 when first encountered.

///

/// Expression | Result | Intuition

/// --------------- | --------- | -----------

/// `NULL AND true` | `NULL` | The `NULL` stands for “unknown” and if it were `true` or `false` the overall expression value could change

/// `NULL AND false` | `false` | If the `NULL` was either `true` or `false` the overall expression is still `false`

/// `NULL AND NULL` | `NULL` |

///

/// Expression | Result | Intuition

/// --------------- | --------- | ----------

/// `NULL OR true` | `true` | If the `NULL` was either `true` or `false` the overall expression is still `true`

/// `NULL OR false` | `NULL` | The `NULL` stands for “unknown” and if it were `true` or `false` the overall expression value could change

/// `NULL OR NULL` | `NULL` |

///

/// ## SQL Filter Semantics

///

/// The SQL `WHERE` clause has a boolean expression, often called a filter or

/// predicate. The semantics of this predicate are that the query evaluates the

/// predicate for each row in the input tables and:

///

/// * Rows that evaluate to `true` are returned in the query results

///

/// * Rows that evaluate to `false` are not returned (“filtered out” or “pruned” or “skipped”).

///

/// * Rows that evaluate to `NULL` are **NOT** returned (also “filtered out”).

/// Note: *this treatment of `NULL` is **DIFFERENT** than how `NULL` is treated

/// in the rewritten predicate described below.*

///

/// # `PruningPredicate` Implementation

///

/// Armed with the information in the Background section, we can now understand

/// how the `PruningPredicate` logic works.

///

/// ## Interface

///

/// **Inputs**

/// 1. An input schema describing what columns exist

///

/// 2. A predicate (expression that evaluates to a boolean)

///

/// 3. [`PruningStatistics`] that provides information about columns in that

/// schema, for multiple “containers”. For each column in each container, it

/// provides optional information on contained values, min_values, max_values,

/// null_counts counts, and row_counts counts.

///

/// **Outputs**:

/// A (non null) boolean value for each container:

/// * `true`: There MAY be rows that match the predicate

///

/// * `false`: There are no rows that could possibly match the predicate (the

/// predicate can never possibly be true). The container can be pruned (skipped)

/// entirely.

///

/// While `PruningPredicate` will never return a `NULL` value, the

/// rewritten predicate (as returned by `build_predicate_expression` and used internally

/// by `PruningPredicate`) may evaluate to `NULL` when some of the min/max values

/// or null / row counts are not known.

///

/// In order to be correct, `PruningPredicate` must return false

/// **only** if it can determine that for all rows in the container, the

/// predicate could never evaluate to `true` (always evaluates to either `NULL`

/// or `false`).

///

/// ## Contains Analysis and Min/Max Rewrite

///

/// `PruningPredicate` works by first analyzing the predicate to see what

/// [`LiteralGuarantee`] must hold for the predicate to be true.

///

/// Then, the `PruningPredicate` rewrites the original predicate into an

/// expression that references the min/max values of each column in the original

/// predicate.

///

/// When the min/max values are actually substituted in to this expression and

/// evaluated, the result means

///

/// * `true`: there MAY be rows that pass the predicate, **KEEPS** the container

///

/// * `NULL`: there MAY be rows that pass the predicate, **KEEPS** the container

/// Note that rewritten predicate can evaluate to NULL when some of

/// the min/max values are not known. *Note that this is different than

/// the SQL filter semantics where `NULL` means the row is filtered

/// out.*

///

/// * `false`: there are no rows that could possibly match the predicate,

/// **PRUNES** the container

///

/// For example, given a column `x`, the `x_min`, `x_max`, `x_null_count`, and

/// `x_row_count` represent the minimum and maximum values, the null count of

/// column `x`, and the row count of column `x`, provided by the `PruningStatistics`.

/// `x_null_count` and `x_row_count` are used to handle the case where the column `x`

/// is known to be all `NULL`s. Note this is different from knowing nothing about

/// the column `x`, which confusingly is encoded by returning `NULL` for the min/max

/// values from [`PruningStatistics::max_values`] and [`PruningStatistics::min_values`].

///

/// Here are some examples of the rewritten predicates:

///

/// Original Predicate | Rewritten Predicate

/// ------------------ | --------------------

/// `x = 5` | `x_null_count != x_row_count AND (x_min <= 5 AND 5 <= x_max)`

/// `x < 5` | `x_null_count != x_row_count AND (x_min < 5)`

/// `x = 5 AND y = 10` | `x_null_count != x_row_count AND (x_min <= 5 AND 5 <= x_max) AND y_null_count != y_row_count (y_min <= 10 AND 10 <= y_max)`

/// `x IS NULL` | `x_null_count > 0`

/// `x IS NOT NULL` | `x_null_count != row_count`

/// `CAST(x as int) = 5` | `x_null_count != x_row_count (CAST(x_min as int) <= 5 AND 5 <= CAST(x_max as int))`

///

/// ## Predicate Evaluation

/// The PruningPredicate works in two passes

///

/// **First pass**: For each `LiteralGuarantee` calls

/// [`PruningStatistics::contained`] and rules out containers where the

/// LiteralGuarantees are not satisfied

///

/// **Second Pass**: Evaluates the rewritten expression using the

/// min/max/null_counts/row_counts values for each column for each container. For any

/// container that this expression evaluates to `false`, it rules out those

/// containers.

///

///

/// ### Example 1

///

/// Given the predicate, `x = 5 AND y = 10`, the rewritten predicate would look like:

///

/// ```sql

/// x_null_count != x_row_count AND (x_min <= 5 AND 5 <= x_max)

/// AND

/// y_null_count != y_row_count AND (y_min <= 10 AND 10 <= y_max)

/// ```

///

/// If we know that for a given container, `x` is between `1 and 100` and we know that

/// `y` is between `4` and `7`, we know nothing about the null count and row count of

/// `x` and `y`, the input statistics might look like:

///

/// Column | Value

/// -------- | -----

/// `x_min` | `1`

/// `x_max` | `100`

/// `x_null_count` | `null`

/// `x_row_count` | `null`

/// `y_min` | `4`

/// `y_max` | `7`

/// `y_null_count` | `null`

/// `y_row_count` | `null`

///

/// When these statistics values are substituted in to the rewritten predicate and

/// simplified, the result is `false`:

///

/// * `null != null AND (1 <= 5 AND 5 <= 100) AND null != null AND (4 <= 10 AND 10 <= 7)`

/// * `null = null` is `null` which is not true, so the AND moves on to the next clause

/// * `null and (1 <= 5 AND 5 <= 100) AND null AND (4 <= 10 AND 10 <= 7)`

/// * evaluating the clauses further we get:

/// * `null and true and null and false`

/// * `null and false`

/// * `false`

///

/// Returning `false` means the container can be pruned, which matches the

/// intuition that `x = 5 AND y = 10` can’t be true for any row if all values of `y`

/// are `7` or less.

///

/// Note that if we had ended up with `null AND true AND null AND true` the result

/// would have been `null`.

/// `null` is treated the same as`true`, because we can't prove that the predicate is `false.`

///

/// If, for some other container, we knew `y` was between the values `4` and

/// `15`, then the rewritten predicate evaluates to `true` (verifying this is

/// left as an exercise to the reader -- are you still here?), and the container

/// **could not** be pruned. The intuition is that there may be rows where the

/// predicate *might* evaluate to `true`, and the only way to find out is to do

/// more analysis, for example by actually reading the data and evaluating the

/// predicate row by row.

///

/// ### Example 2

///

/// Given the same predicate, `x = 5 AND y = 10`, the rewritten predicate would

/// look like the same as example 1:

///

/// ```sql

/// x_null_count != x_row_count AND (x_min <= 5 AND 5 <= x_max)

/// AND

/// y_null_count != y_row_count AND (y_min <= 10 AND 10 <= y_max)

/// ```

///

/// If we know that for another given container, `x_min` is NULL and `x_max` is

/// NULL (the min/max values are unknown), `x_null_count` is `100` and `x_row_count`

/// is `100`; we know that `y` is between `4` and `7`, but we know nothing about

/// the null count and row count of `y`. The input statistics might look like:

///

/// Column | Value

/// -------- | -----

/// `x_min` | `null`

/// `x_max` | `null`

/// `x_null_count` | `100`

/// `x_row_count` | `100`

/// `y_min` | `4`

/// `y_max` | `7`

/// `y_null_count` | `null`

/// `y_row_count` | `null`

///

/// When these statistics values are substituted in to the rewritten predicate and

/// simplified, the result is `false`:

///

/// * `100 != 100 AND (null <= 5 AND 5 <= null) AND null = null AND (4 <= 10 AND 10 <= 7)`

/// * `false AND null AND null AND false`

/// * `false AND false`

/// * `false`

///

/// Returning `false` means the container can be pruned, which matches the

/// intuition that `x = 5 AND y = 10` can’t be true because all values in `x`

/// are known to be NULL.

///

/// # Related Work

///

/// [`PruningPredicate`] implements the type of min/max pruning described in

/// Section `3.3.3` of the [`Snowflake SIGMOD Paper`]. The technique is

/// described by various research such as [small materialized aggregates], [zone

/// maps], and [data skipping].

///

/// [`Snowflake SIGMOD Paper`]: https://dl.acm.org/doi/10.1145/2882903.2903741

/// [small materialized aggregates]: https://www.vldb.org/conf/1998/p476.pdf

/// [zone maps]: https://dl.acm.org/doi/10.1007/978-3-642-03730-6_10

/// [data skipping]: https://dl.acm.org/doi/10.1145/2588555.2610515

#[derive(Debug, Clone)]

pub struct PruningPredicate {

/// The input schema against which the predicate will be evaluated

schema: SchemaRef,

/// A min/max pruning predicate (rewritten in terms of column min/max

/// values, which are supplied by statistics)

predicate_expr: Arc<dyn PhysicalExpr>,

/// Description of which statistics are required to evaluate `predicate_expr`

required_columns: RequiredColumns,

/// Original physical predicate from which this predicate expr is derived

/// (required for serialization)

orig_expr: Arc<dyn PhysicalExpr>,

/// [`LiteralGuarantee`]s used to try and prove a predicate can not possibly

/// evaluate to `true`.

///

/// See [`PruningPredicate::literal_guarantees`] for more details.

literal_guarantees: Vec<LiteralGuarantee>,

/// Maximum IN-list size used to build this predicate.

max_in_list_size: usize,

/// Whether its logical inverse can safely prove every row matches.

can_be_inverted_for_full_match: bool,

}

#[derive(Default)]

struct PruningExpressionProperties {

/// The rewrite preserves which rows can be TRUE, but not whether rejected

/// rows are FALSE or UNKNOWN.

has_filter_semantics_only: bool,

}

/// Build a pruning predicate from an optional predicate expression.

/// If the predicate is None or the predicate cannot be converted to a pruning

/// predicate, return None.

/// If there is an error creating the pruning predicate it is recorded by incrementing

/// the `predicate_creation_errors` counter.

pub fn build_pruning_predicate(

predicate: Arc<dyn PhysicalExpr>,

file_schema: &SchemaRef,

predicate_creation_errors: &Count,

) -> Option<Arc<PruningPredicate>> {

PruningPredicateBuilder::new()

.with_file_schema(Arc::clone(file_schema))

.with_error_counter(predicate_creation_errors)

.build(predicate)

}

/// Builder for a [`PruningPredicate`]. Groups optional configuration —

/// `IN (...)` rewrite cap, error counter — so future additions do not

/// churn the top-level API.

///

/// The two entry points are:

/// - [`Self::build`]: convenience for scan sites that already track a

/// `predicate_creation_errors` counter. Returns `Some(Arc<..>)` when the

/// resulting predicate can actually prune, `None` when it is trivially

/// true or when construction failed (in which case the error counter is

/// incremented if one was supplied).

/// - [`Self::try_build`]: returns a raw `Result<PruningPredicate>` for

/// callers that want to surface errors themselves.

///

#[derive(Default)]

pub struct PruningPredicateBuilder<'a> {

file_schema: Option<SchemaRef>,

error_counter: Option<&'a Count>,

max_in_list_size: usize,

}

impl<'a> PruningPredicateBuilder<'a> {

/// Create a new builder with the default pruning predicate configuration.

pub fn new() -> Self {

Self {

file_schema: None,

error_counter: None,

max_in_list_size: MAX_IN_LIST_SIZE,

}

}

/// Set the schema of the container that will be pruned (typically the

/// parquet file schema).

pub fn with_file_schema(mut self, file_schema: SchemaRef) -> Self {

self.file_schema = Some(file_schema);

self

}

/// Metric counter incremented once per predicate that fails to build.

/// Only consulted by [`Self::build`]; [`Self::try_build`] surfaces the

/// error directly.

pub fn with_error_counter(mut self, error_counter: &'a Count) -> Self {

self.error_counter = Some(error_counter);

self

}

/// Cap on the input length of `IN (...)` lists eligible for min/max pruning.

///

/// For a nonempty list of `N` values (before deduplication) and cap `C`:

///

/// | Condition | Pruning representation |

/// | --- | --- |

/// | `N <= min(20, C)` | Existing per-value rewrite |

/// | `20 < N <= C`, supported ordered literals with optional NULLs | Compact pruning expression |

/// | `20 < N <= C`, unsupported lists | Existing per-value rewrite |

/// | `N > C` | Unhandled-predicate hook, normally "keep the container" |

///

/// Empty lists also use the unhandled-predicate hook. A cap of zero disables

/// this `IN` min/max rewrite, not other predicates or literal/containment

/// pruning (such as Bloom filters). The default cap is [`MAX_IN_LIST_SIZE`]

/// (20), so the compact path requires an explicitly raised cap.

///

/// Compact domains support strings, binary values, integers, decimals,

/// dates, times, timestamps, and durations. Floating-point values are

/// excluded because NaN and signed zero do not follow the required order.

///

/// The compact form covers `IN` and `NOT IN` alike, including lists with

/// NULL members. Raising the cap can still build large comparison trees for

/// unsupported lists.

///

/// Query engines typically pass

/// `datafusion.execution.parquet.max_in_list_size` here.

pub fn with_max_in_list_size(mut self, max_in_list_size: usize) -> Self {

self.max_in_list_size = max_in_list_size;

self

}

/// Build a [`PruningPredicate`] wrapped in `Some(Arc<..>)` when it can

/// prune, `None` when it is trivially true or when construction fails.

/// If [`Self::with_error_counter`] was set, construction failures are

/// recorded there.

pub fn build(

self,

predicate: Arc<dyn PhysicalExpr>,

) -> Option<Arc<PruningPredicate>> {

let error_counter = self.error_counter;

match self.try_build(predicate) {

Ok(pruning_predicate) => {

if !pruning_predicate.always_true() {

return Some(Arc::new(pruning_predicate));

}

}

Err(e) => {

debug!("Could not create pruning predicate for: {e}");

if let Some(counter) = error_counter {

counter.add(1);

}

}

}

None

}

/// Build a [`PruningPredicate`], returning the construction error

/// directly. Callers that want the always-true predicate elided or

/// errors folded into a counter should use [`Self::build`] instead.

pub fn try_build(

self,

mut predicate: Arc<dyn PhysicalExpr>,

) -> Result<PruningPredicate> {

let file_schema = self.file_schema.ok_or_else(|| {

_internal_datafusion_err!(

"PruningPredicateBuilder requires a file schema (call `with_file_schema`)"

)

})?;

// Get a (simpler) snapshot of the physical expr here to use with `PruningPredicate`.

// In particular this unravels any `DynamicFilterPhysicalExpr`s by snapshotting them

// so that PruningPredicate can work with a static expression.

let tf = snapshot_physical_expr_opt(predicate)?;

if tf.transformed {

// If we had an expression such as Dynamic(part_col < 5 and col < 10)

// (this could come from something like `select * from t order by part_col, col, limit 10`)

// after snapshotting and because `DynamicFilterPhysicalExpr` applies child replacements to its

// children after snapshotting and previously `replace_columns_with_literals` may have been called with partition values

// the expression we have now is `8 < 5 and col < 10`.

// Thus we need as simplifier pass to get `false and col < 10` => `false` here.

let simplifier = PhysicalExprSimplifier::new(&file_schema);

predicate = simplifier.simplify(tf.data)?;

} else {

predicate = tf.data;

}

let unhandled_hook = Arc::new(ConstantUnhandledPredicateHook::default()) as _;

// build predicate expression once

let mut required_columns = RequiredColumns::new();

let mut properties = PruningExpressionProperties::default();

let predicate_expr = build_predicate_expression(

&predicate,

&file_schema,

&mut required_columns,

&unhandled_hook,

self.max_in_list_size,

&mut properties,

);

let predicate_schema = required_columns.schema();

// Simplify the newly created predicate to get rid of redundant casts, comparisons, etc.

let predicate_expr =

PhysicalExprSimplifier::new(&predicate_schema).simplify(predicate_expr)?;

let literal_guarantees = LiteralGuarantee::analyze(&predicate);

Ok(PruningPredicate {

schema: file_schema,

predicate_expr,

required_columns,

orig_expr: predicate,

literal_guarantees,

max_in_list_size: self.max_in_list_size,

can_be_inverted_for_full_match: !properties.has_filter_semantics_only,

})

}

}

/// Rewrites predicates that [`PredicateRewriter`] can not handle, e.g. certain

/// complex expressions or predicates that reference columns that are not in the

/// schema.

pub trait UnhandledPredicateHook {

/// Called when a predicate can not be rewritten in terms of statistics or

/// references a column that is not in the schema.

fn handle(&self, expr: &Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr>;

}

/// The default handling for unhandled predicates is to return a constant `true`

/// (meaning don't prune the container)

#[derive(Debug, Clone)]

struct ConstantUnhandledPredicateHook {

default: Arc<dyn PhysicalExpr>,

}

impl Default for ConstantUnhandledPredicateHook {

fn default() -> Self {

Self {

default: Arc::new(phys_expr::Literal::new(ScalarValue::from(true))),

}

}

}

impl UnhandledPredicateHook for ConstantUnhandledPredicateHook {

fn handle(&self, _expr: &Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {

Arc::clone(&self.default)

}

}

impl PruningPredicate {

/// Try to create a new instance of [`PruningPredicate`]

///

/// This will translate the provided `expr` filter expression into

/// a *pruning predicate*.

///

/// A pruning predicate is one that has been rewritten in terms of

/// the min and max values of column references and that evaluates

/// to FALSE if the filter predicate would evaluate FALSE *for

/// every row* whose values fell within the min / max ranges (aka

/// could be pruned).

///

/// The pruning predicate evaluates to TRUE or NULL

/// if the filter predicate *might* evaluate to TRUE for at least

/// one row whose values fell within the min/max ranges (in other

/// words they might pass the predicate)

///

/// For example, the filter expression `(column / 2) = 4` becomes

/// the pruning predicate

/// `(column_min / 2) <= 4 && 4 <= (column_max / 2))`

///

/// See the struct level documentation on [`PruningPredicate`] for more

/// details.

///

/// Note that `PruningPredicate` does not attempt to normalize or simplify

/// the input expression unless calling [`snapshot_physical_expr_opt`]

/// returns a new expression.

/// It is recommended that you pass the expressions through [`PhysicalExprSimplifier`]

/// before calling this method to make sure the expressions can be used for pruning.

///

/// Use [`PruningPredicateBuilder`] to construct new pruning predicates.

#[deprecated(since = "55.0.0", note = "Use PruningPredicateBuilder instead")]

pub fn try_new(expr: Arc<dyn PhysicalExpr>, schema: SchemaRef) -> Result<Self> {

PruningPredicateBuilder::new()

.with_file_schema(schema)

.try_build(expr)

}

/// For each set of statistics, evaluates the pruning predicate

/// and returns a `bool` with the following meaning for a

/// all rows whose values match the statistics:

///

/// `true`: There MAY be rows that match the predicate

///

/// `false`: There are no rows that could possibly match the predicate

///

/// Note: the predicate passed to `prune` should already be simplified as

/// much as possible (e.g. this pass doesn't handle some

/// expressions like `b = false`, but it does handle the

/// simplified version `b`. See [`ExprSimplifier`] to simplify expressions.

///

/// [`ExprSimplifier`]: https://docs.rs/datafusion/latest/datafusion/optimizer/simplify_expressions/struct.ExprSimplifier.html

pub fn prune<S: PruningStatistics + ?Sized>(

&self,

statistics: &S,

) -> Result<Vec<bool>> {

let mut builder = BoolVecBuilder::new(statistics.num_containers());

// Try to prove the predicate can't be true for the containers based on

// literal guarantees

for literal_guarantee in &self.literal_guarantees {

let LiteralGuarantee {

column,

guarantee,

literals,

} = literal_guarantee;

if let Some(results) = statistics.contained(column, literals) {

match guarantee {

// `In` means the values in the column must be one of the

// values in the set for the predicate to evaluate to true.

// If `contained` returns false, that means the column is

// not any of the values so we can prune the container

Guarantee::In => builder.combine_array(&results),

// `NotIn` means the values in the column must not be

// any of the values in the set for the predicate to

// evaluate to true. If `contained` returns true, it means the

// column is only in the set of values so we can prune the

// container

Guarantee::NotIn => {

builder.combine_array(&arrow::compute::not(&results)?)

}

}

// if all containers are pruned (has rows that DEFINITELY DO NOT pass the predicate)

// can return early without evaluating the rest of predicates.

if builder.check_all_pruned() {

return Ok(builder.build());

}

}

}

// Next, try to prove the predicate can't be true for the containers based

// on min/max values

// build a RecordBatch that contains the min/max values in the

// appropriate statistics columns for the min/max predicate

let statistics_batch =

build_statistics_record_batch(statistics, &self.required_columns)?;

// Evaluate the pruning predicate on that record batch and append any results to the builder

builder.combine_value(self.predicate_expr.evaluate(&statistics_batch)?);

Ok(builder.build())

}

/// Return a reference to the input schema

pub fn schema(&self) -> &SchemaRef {

&self.schema

}

/// Returns a reference to the physical expr used to construct this pruning predicate

pub fn orig_expr(&self) -> &Arc<dyn PhysicalExpr> {

&self.orig_expr

}

/// Returns a reference to the predicate expr

pub fn predicate_expr(&self) -> &Arc<dyn PhysicalExpr> {

&self.predicate_expr

}

/// Returns a reference to the literal guarantees

///

/// Note that **All** `LiteralGuarantee`s must be satisfied for the

/// expression to possibly be `true`. If any is not satisfied, the

/// expression is guaranteed to be `null` or `false`.

pub fn literal_guarantees(&self) -> &[LiteralGuarantee] {

&self.literal_guarantees

}

/// Returns true if this pruning predicate can not prune anything.

///

/// This happens if the predicate is a literal `true` and

/// literal_guarantees is empty.

///

/// This can happen when a predicate is simplified to a constant `true`

pub fn always_true(&self) -> bool {

is_always_true(&self.predicate_expr) && self.literal_guarantees.is_empty()

}

/// Returns the configured maximum IN-list size used to build this predicate.

pub fn max_in_list_size(&self) -> usize {

self.max_in_list_size

}

/// Returns whether pruning the logical inverse can safely prove that every

/// row in a container satisfies the original predicate.

pub fn can_be_inverted_for_full_match(&self) -> bool {

self.can_be_inverted_for_full_match

}

pub fn required_columns(&self) -> &RequiredColumns {

&self.required_columns

}

/// Names of the columns that are known to be / not be in a set

/// of literals (constants). These are the columns the that may be passed to

/// [`PruningStatistics::contained`] during pruning.

///

/// This is useful to avoid fetching statistics for columns that will not be

/// used in the predicate. For example, it can be used to avoid reading

/// unneeded bloom filters (a non trivial operation).

pub fn literal_columns(&self) -> Vec<String> {

let mut seen = HashSet::with_capacity(self.literal_guarantees.len());

self.literal_guarantees

.iter()

.map(|e| &e.column.name)

// avoid duplicates

.filter(|name| seen.insert(*name))

.map(|s| s.to_string())

.collect()

}

}

/// Builds the return `Vec` for [`PruningPredicate::prune`].

#[derive(Debug)]

struct BoolVecBuilder {

/// One element per container. Each element is

/// * `true`: if the container has row that may pass the predicate

/// * `false`: if the container has rows that DEFINITELY DO NOT pass the predicate

inner: Vec<bool>,

}

impl BoolVecBuilder {

/// Create a new `BoolVecBuilder` with `num_containers` elements

fn new(num_containers: usize) -> Self {

Self {

// assume by default all containers may pass the predicate

inner: vec![true; num_containers],

}

}

/// Combines result `array` for a conjunct (e.g. `AND` clause) of a

/// predicate into the currently in progress array.

///

/// Each `array` element is:

/// * `true`: container has row that may pass the predicate

/// * `false`: all container rows DEFINITELY DO NOT pass the predicate

/// * `null`: container may or may not have rows that pass the predicate

fn combine_array(&mut self, array: &BooleanArray) {

assert_eq!(array.len(), self.inner.len());

for (cur, new) in self.inner.iter_mut().zip(array.iter()) {

// `false` for this conjunct means we know for sure no rows could

// pass the predicate and thus we set the corresponding container

// location to false.

if new == Some(false) {

*cur = false;

}

}

}

/// Combines the results in the [`ColumnarValue`] to the currently in

/// progress array, following the same rules as [`Self::combine_array`].

///

/// # Panics

/// If `value` is not boolean

fn combine_value(&mut self, value: ColumnarValue) {

match value {

ColumnarValue::Array(array) => {

self.combine_array(array.as_boolean());

}

ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))) => {

// False means all containers can not pass the predicate

self.inner = vec![false; self.inner.len()];

}

_ => {

// Null or true means the rows in container may pass this

// conjunct so we can't prune any containers based on that

}

}

}

/// Convert this builder into a Vec of bools

fn build(self) -> Vec<bool> {

self.inner

}

/// Check all containers has rows that DEFINITELY DO NOT pass the predicate

fn check_all_pruned(&self) -> bool {

self.inner.iter().all(|&x| !x)

}

}

fn is_always_true(expr: &Arc<dyn PhysicalExpr>) -> bool {

expr.downcast_ref::<phys_expr::Literal>()

.is_some_and(|l| *l.value() == ScalarValue::Boolean(Some(true)))

}

fn is_always_false(expr: &Arc<dyn PhysicalExpr>) -> bool {

expr.downcast_ref::<phys_expr::Literal>()

.is_some_and(|l| *l.value() == ScalarValue::Boolean(Some(false)))

}

/// Describes which columns statistics are necessary to evaluate a

/// [`PruningPredicate`].

///

/// This structure permits reading and creating the minimum number statistics,

/// which is important since statistics may be non trivial to read (e.g. large

/// strings or when there are 1000s of columns).

///

/// Handles creating references to the min/max statistics

/// for columns as well as recording which statistics are needed

#[derive(Debug, Default, Clone)]

pub struct RequiredColumns {

/// The statistics required to evaluate this predicate:

/// * The unqualified column in the input schema

/// * Statistics type (e.g. Min or Max or Null_Count)

/// * The field the statistics value should be placed in for

/// pruning predicate evaluation (e.g. `min_value` or `max_value`)

columns: Vec<(phys_expr::Column, StatisticsType, Field)>,

}

impl RequiredColumns {

fn new() -> Self {

Self::default()

}

/// Returns Some(column) if this is a single column predicate.

///

/// Returns None if this is a multi-column predicate.

///

/// Examples:

/// * `a > 5 OR a < 10` returns `Some(a)`

/// * `a > 5 OR b < 10` returns `None`

/// * `true` returns None

pub fn single_column(&self) -> Option<&phys_expr::Column> {

if self.columns.windows(2).all(|w| {

// check if all columns are the same (ignoring statistics and field)

let c1 = &w[0].0;

let c2 = &w[1].0;

c1 == c2

}) {

self.columns.first().map(|r| &r.0)

} else {

None

}

}

/// Returns a schema that describes the columns required to evaluate this

/// pruning predicate.

/// The schema contains the fields for each column in `self.columns` with

/// the appropriate data type for the statistics.

/// Order matters, this same order is used to evaluate the

/// pruning predicate.

fn schema(&self) -> Schema {

let fields = self

.columns

.iter()

.map(|(_c, _t, f)| f.clone())

.collect::<Vec<_>>();

Schema::new(fields)

}

/// Returns an iterator over items in columns (see doc on

/// `self.columns` for details)

pub(crate) fn iter(

&self,

) -> impl Iterator<Item = &(phys_expr::Column, StatisticsType, Field)> {

self.columns.iter()

}

fn find_stat_column(

&self,

column: &phys_expr::Column,

statistics_type: StatisticsType,

) -> Option<usize> {

match statistics_type {

StatisticsType::RowCount => {

// Use the first row count we find, if any

self.columns

.iter()

.enumerate()

.find(|(_i, (_c, t, _f))| t == &statistics_type)

.map(|(i, (_c, _t, _f))| i)

}

_ => self

.columns

.iter()

.enumerate()

.find(|(_i, (c, t, _f))| c == column && t == &statistics_type)

.map(|(i, (_c, _t, _f))| i),

}

}

/// Rewrites column_expr so that all appearances of column

/// are replaced with a reference to either the min or max

/// statistics column, while keeping track that a reference to the statistics

/// column is required

///

/// for example, an expression like `col("foo") > 5`, when called

/// with Max would result in an expression like `col("foo_max") >

/// 5` with the appropriate entry noted in self.columns

fn stat_column_expr(

&mut self,

column: &phys_expr::Column,

column_expr: &Arc<dyn PhysicalExpr>,

field: &Field,

stat_type: StatisticsType,

) -> Result<Arc<dyn PhysicalExpr>> {

let (idx, need_to_insert) = match self.find_stat_column(column, stat_type) {

Some(idx) => (idx, false),

None => (self.columns.len(), true),

};

let column_name = column.name();

let stat_column_name = match stat_type {

StatisticsType::Min => format!("{column_name}_min"),

StatisticsType::Max => format!("{column_name}_max"),

StatisticsType::NullCount => format!("{column_name}_null_count"),

StatisticsType::RowCount => "row_count".to_string(),

};

let stat_column = phys_expr::Column::new(&stat_column_name, idx);

// only add statistics column if not previously added

if need_to_insert {

// may be null if statistics are not present

let nullable = true;

let stat_field =

Field::new(stat_column.name(), field.data_type().clone(), nullable);

self.columns.push((column.clone(), stat_type, stat_field));

}

rewrite_column_expr(Arc::clone(column_expr), column, &stat_column)

}

/// rewrite col --> col_min

fn min_column_expr(

&mut self,

column: &phys_expr::Column,

column_expr: &Arc<dyn PhysicalExpr>,

field: &Field,

) -> Result<Arc<dyn PhysicalExpr>> {

self.stat_column_expr(column, column_expr, field, StatisticsType::Min)

}

/// rewrite col --> col_max

fn max_column_expr(

&mut self,

column: &phys_expr::Column,

column_expr: &Arc<dyn PhysicalExpr>,

field: &Field,

) -> Result<Arc<dyn PhysicalExpr>> {

self.stat_column_expr(column, column_expr, field, StatisticsType::Max)

}

/// rewrite col --> col_null_count

fn null_count_column_expr(

&mut self,

column: &phys_expr::Column,

column_expr: &Arc<dyn PhysicalExpr>,

field: &Field,