datafusion_pruning/

pruning_predicate.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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.

//! [`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 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::{Column, DFSchema, 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::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::{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 [`PruningPredicate::try_new`] 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>,
}

/// 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>> {
    match PruningPredicate::try_new(predicate, Arc::clone(file_schema)) {
        Ok(pruning_predicate) => {
            if !pruning_predicate.always_true() {
                return Some(Arc::new(pruning_predicate));
            }
        }
        Err(e) => {
            debug!("Could not create pruning predicate for: {e}");
            predicate_creation_errors.add(1);
        }
    }
    None
}

/// 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.
    pub fn try_new(mut expr: Arc<dyn PhysicalExpr>, schema: SchemaRef) -> Result<Self> {
        // 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(expr)?;
        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(&schema);
            expr = simplifier.simplify(tf.data)?;
        } else {
            expr = tf.data;
        }
        let unhandled_hook = Arc::new(ConstantUnhandledPredicateHook::default()) as _;

        // build predicate expression once
        let mut required_columns = RequiredColumns::new();
        let predicate_expr = build_predicate_expression(
            &expr,
            &schema,
            &mut required_columns,
            &unhandled_hook,
        );
        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(&expr);

        Ok(Self {
            schema,
            predicate_expr,
            required_columns,
            orig_expr: expr,
            literal_guarantees,
        })
    }

    /// 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()
    }

    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::new();
        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 let Some(false) = new {
                *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>()
        .map(|l| matches!(l.value(), ScalarValue::Boolean(Some(true))))
        .unwrap_or_default()
}

fn is_always_false(expr: &Arc<dyn PhysicalExpr>) -> bool {
    expr.downcast_ref::<phys_expr::Literal>()
        .map(|l| matches!(l.value(), ScalarValue::Boolean(Some(false))))
        .unwrap_or_default()
}

/// 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,
    ) -> Result<Arc<dyn PhysicalExpr>> {
        self.stat_column_expr(column, column_expr, field, StatisticsType::NullCount)
    }

    /// rewrite col --> col_row_count
    fn row_count_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::RowCount)
    }
}

impl From<Vec<(phys_expr::Column, StatisticsType, Field)>> for RequiredColumns {
    fn from(columns: Vec<(phys_expr::Column, StatisticsType, Field)>) -> Self {
        Self { columns }
    }
}

/// Build a RecordBatch from a list of statistics, creating arrays,
/// with one row for each PruningStatistics and columns specified in
/// the required_columns parameter.
///
/// For example, if the requested columns are
/// ```text
/// ("s1", Min, Field:s1_min)
/// ("s2", Max, field:s2_max)
/// ```
///
/// And the input statistics had
/// ```text
/// S1(Min: 5, Max: 10)
/// S2(Min: 99, Max: 1000)
/// S3(Min: 1, Max: 2)
/// ```
///
/// Then this function would build a record batch with 2 columns and
/// one row s1_min and s2_max as follows (s3 is not requested):
///
/// ```text
/// s1_min | s2_max
/// -------+--------
///   5    | 1000
/// ```
fn build_statistics_record_batch<S: PruningStatistics + ?Sized>(
    statistics: &S,
    required_columns: &RequiredColumns,
) -> Result<RecordBatch> {
    let mut arrays = Vec::<ArrayRef>::new();
    // For each needed statistics column:
    for (column, statistics_type, stat_field) in required_columns.iter() {
        let column = Column::from_name(column.name());
        let data_type = stat_field.data_type();

        let num_containers = statistics.num_containers();

        let array = match statistics_type {
            StatisticsType::Min => statistics.min_values(&column),
            StatisticsType::Max => statistics.max_values(&column),
            StatisticsType::NullCount => statistics.null_counts(&column),
            StatisticsType::RowCount => statistics.row_counts(),
        };
        let array = array.unwrap_or_else(|| new_null_array(data_type, num_containers));

        assert_eq_or_internal_err!(
            num_containers,
            array.len(),
            "mismatched statistics length. Expected {}, got {}",
            num_containers,
            array.len()
        );

        // cast statistics array to required data type (e.g. parquet
        // provides timestamp statistics as "Int64")
        let array = arrow::compute::cast(&array, data_type)?;

        arrays.push(array);
    }

    let schema = Arc::new(required_columns.schema());
    // provide the count in case there were no needed statistics
    let mut options = RecordBatchOptions::default();
    options.row_count = Some(statistics.num_containers());

    trace!("Creating statistics batch for {required_columns:#?} with {arrays:#?}");

    RecordBatch::try_new_with_options(schema, arrays, &options).map_err(|err| {
        plan_datafusion_err!("Can not create statistics record batch: {err}")
    })
}

struct PruningExpressionBuilder<'a> {
    column: phys_expr::Column,
    column_expr: Arc<dyn PhysicalExpr>,
    op: Operator,
    scalar_expr: Arc<dyn PhysicalExpr>,
    field: &'a Field,
    required_columns: &'a mut RequiredColumns,
}

impl<'a> PruningExpressionBuilder<'a> {
    fn try_new(
        left: &'a Arc<dyn PhysicalExpr>,
        right: &'a Arc<dyn PhysicalExpr>,
        left_columns: ColumnReferenceCount,
        right_columns: ColumnReferenceCount,
        op: Operator,
        schema: &'a SchemaRef,
        required_columns: &'a mut RequiredColumns,
    ) -> Result<Self> {
        // find column name; input could be a more complicated expression
        let (column_expr, scalar_expr, column, correct_operator) = match (
            left_columns,
            right_columns,
        ) {
            (ColumnReferenceCount::One(column), ColumnReferenceCount::Zero) => {
                (left, right, column, op)
            }
            (ColumnReferenceCount::Zero, ColumnReferenceCount::One(column)) => {
                (right, left, column, reverse_operator(op)?)
            }
            (ColumnReferenceCount::One(_), ColumnReferenceCount::One(_)) => {
                // both sides have one column - not supported
                return plan_err!(
                    "Expression not supported for pruning: left has 1 column, right has 1 column"
                );
            }
            (ColumnReferenceCount::Zero, ColumnReferenceCount::Zero) => {
                // both sides are literals - should be handled before calling try_new
                return plan_err!(
                    "Pruning literal expressions is not supported, please call PhysicalExprSimplifier first"
                );
            }
            (ColumnReferenceCount::Many, _) | (_, ColumnReferenceCount::Many) => {
                return plan_err!(
                    "Expression not supported for pruning: left or right has multiple columns"
                );
            }
        };

        let df_schema = DFSchema::try_from(Arc::clone(schema))?;
        let (column_expr, correct_operator, scalar_expr) = rewrite_expr_to_prunable(
            column_expr,
            correct_operator,
            scalar_expr,
            df_schema,
        )?;
        let field = match schema.column_with_name(column.name()) {
            Some((_, f)) => f,
            _ => {
                return plan_err!("Field not found in schema");
            }
        };

        Ok(Self {
            column,
            column_expr,
            op: correct_operator,
            scalar_expr,
            field,
            required_columns,
        })
    }

    fn op(&self) -> Operator {
        self.op
    }

    fn scalar_expr(&self) -> &Arc<dyn PhysicalExpr> {
        &self.scalar_expr
    }

    fn min_column_expr(&mut self) -> Result<Arc<dyn PhysicalExpr>> {
        self.required_columns
            .min_column_expr(&self.column, &self.column_expr, self.field)
    }

    fn max_column_expr(&mut self) -> Result<Arc<dyn PhysicalExpr>> {
        self.required_columns
            .max_column_expr(&self.column, &self.column_expr, self.field)
    }

    /// This function is to simply retune the `null_count` physical expression no matter what the
    /// predicate expression is
    ///
    /// i.e., x > 5 => x_null_count,
    ///       cast(x as int) < 10 => x_null_count,
    ///       try_cast(x as float) < 10.0 => x_null_count
    fn null_count_column_expr(&mut self) -> Result<Arc<dyn PhysicalExpr>> {
        // Retune to [`phys_expr::Column`]
        let column_expr = Arc::new(self.column.clone()) as _;

        // null_count is DataType::UInt64, which is different from the column's data type (i.e. self.field)
        let null_count_field = &Field::new(self.field.name(), DataType::UInt64, true);

        self.required_columns.null_count_column_expr(
            &self.column,
            &column_expr,
            null_count_field,
        )
    }

    /// This function is to simply retune the `row_count` physical expression no matter what the
    /// predicate expression is
    ///
    /// i.e., x > 5 => x_row_count,
    ///       cast(x as int) < 10 => x_row_count,
    ///       try_cast(x as float) < 10.0 => x_row_count
    fn row_count_column_expr(&mut self) -> Result<Arc<dyn PhysicalExpr>> {
        // Retune to [`phys_expr::Column`]
        let column_expr = Arc::new(self.column.clone()) as _;

        // row_count is DataType::UInt64, which is different from the column's data type (i.e. self.field)
        let row_count_field = &Field::new(self.field.name(), DataType::UInt64, true);

        self.required_columns.row_count_column_expr(
            &self.column,
            &column_expr,
            row_count_field,
        )
    }
}

/// This function is designed to rewrite the column_expr to
/// ensure the column_expr is monotonically increasing.
///
/// For example,
/// 1. `col > 10`
/// 2. `-col > 10` should be rewritten to `col < -10`
/// 3. `!col = true` would be rewritten to `col = !true`
/// 4. `abs(a - 10) > 0` not supported
/// 5. `cast(can_prunable_expr) > 10`
/// 6. `try_cast(can_prunable_expr) > 10`
///
/// More rewrite rules are still in progress.
fn rewrite_expr_to_prunable(
    column_expr: &PhysicalExprRef,
    op: Operator,
    scalar_expr: &PhysicalExprRef,
    schema: DFSchema,
) -> Result<(PhysicalExprRef, Operator, PhysicalExprRef)> {
    if !is_compare_op(op) {
        return plan_err!("rewrite_expr_to_prunable only support compare expression");
    }

    if column_expr.downcast_ref::<phys_expr::Column>().is_some() {
        // `col op lit()`
        Ok((Arc::clone(column_expr), op, Arc::clone(scalar_expr)))
    } else if let Some(cast) = column_expr.downcast_ref::<phys_expr::CastExpr>() {
        // `cast(col) op lit()`
        let (left, op, right) = rewrite_cast_child_to_prunable(
            cast.expr(),
            cast.cast_type(),
            op,
            scalar_expr,
            schema,
        )?;
        let left = Arc::new(phys_expr::CastExpr::new_with_target_field(
            left,
            Arc::clone(cast.target_field()),
            None,
        ));
        // PruningPredicate does not support pruning on nested fields yet.
        // End-to-end nested-field pruning also requires Parquet statistics
        // extraction to agree with PruningPredicate on a stats representation
        // for nested field expressions.
        Ok((left, op, right))
    } else if let Some(try_cast) = column_expr.downcast_ref::<phys_expr::TryCastExpr>() {
        // `try_cast(col) op lit()`
        let (left, op, right) = rewrite_cast_child_to_prunable(
            try_cast.expr(),
            try_cast.cast_type(),
            op,
            scalar_expr,
            schema,
        )?;
        let left = Arc::new(phys_expr::TryCastExpr::new(
            left,
            try_cast.cast_type().clone(),
        ));
        Ok((left, op, right))
    } else if let Some(neg) = column_expr.downcast_ref::<phys_expr::NegativeExpr>() {
        // `-col > lit()`  --> `col < -lit()`
        let (left, op, right) =
            rewrite_expr_to_prunable(neg.arg(), op, scalar_expr, schema)?;
        let right = Arc::new(phys_expr::NegativeExpr::new(right));
        Ok((left, reverse_operator(op)?, right))
    } else if let Some(not) = column_expr.downcast_ref::<phys_expr::NotExpr>() {
        // `!col = true` --> `col = !true`
        if !matches!(
            op,
            Operator::Eq
                | Operator::NotEq
                | Operator::IsDistinctFrom
                | Operator::IsNotDistinctFrom
        ) {
            return plan_err!(
                "Not with operator other than Eq / NotEq / IsDistinctFrom / IsNotDistinctFrom is not supported"
            );
        }
        if not.arg().downcast_ref::<phys_expr::Column>().is_some() {
            let left = Arc::clone(not.arg());
            let right = Arc::new(phys_expr::NotExpr::new(Arc::clone(scalar_expr)));
            Ok((left, reverse_operator(op)?, right))
        } else {
            plan_err!("Not with complex expression {column_expr:?} is not supported")
        }
    } else {
        plan_err!("column expression {column_expr:?} is not supported")
    }
}

fn rewrite_cast_child_to_prunable(
    cast_child_expr: &PhysicalExprRef,
    cast_type: &DataType,
    op: Operator,
    scalar_expr: &PhysicalExprRef,
    schema: DFSchema,
) -> Result<(PhysicalExprRef, Operator, PhysicalExprRef)> {
    verify_support_type_for_prune(
        &cast_child_expr.data_type(schema.as_arrow())?,
        cast_type,
    )?;
    rewrite_expr_to_prunable(cast_child_expr, op, scalar_expr, schema)
}

fn is_compare_op(op: Operator) -> bool {
    matches!(
        op,
        Operator::Eq
            | Operator::NotEq
            | Operator::Lt
            | Operator::LtEq
            | Operator::Gt
            | Operator::GtEq
            | Operator::IsDistinctFrom
            | Operator::IsNotDistinctFrom
            | Operator::LikeMatch
            | Operator::NotLikeMatch
    )
}

// The pruning logic is based on the comparing the min/max bounds.
// Must make sure the two type has order.
// For example, casts from string to numbers is not correct.
// Because the "13" is less than "3" with UTF8 comparison order.
fn verify_support_type_for_prune(from_type: &DataType, to_type: &DataType) -> Result<()> {
    // Dictionary casts are always supported as long as the value types are supported
    let from_type = match from_type {
        DataType::Dictionary(_, t) => {
            return verify_support_type_for_prune(t.as_ref(), to_type);
        }
        _ => from_type,
    };
    let to_type = match to_type {
        DataType::Dictionary(_, t) => {
            return verify_support_type_for_prune(from_type, t.as_ref());
        }
        _ => to_type,
    };
    // If both types are strings or both are not strings (number, timestamp, etc)
    // then we can compare them.
    // PruningPredicate does not support casting of strings to numbers and such.
    if from_type.is_string() == to_type.is_string() {
        Ok(())
    } else {
        plan_err!(
            "Try Cast/Cast with from type {from_type} to type {to_type} is not supported"
        )
    }
}

/// replaces a column with an old name with a new name in an expression
fn rewrite_column_expr(
    e: Arc<dyn PhysicalExpr>,
    column_old: &phys_expr::Column,
    column_new: &phys_expr::Column,
) -> Result<Arc<dyn PhysicalExpr>> {
    e.transform(|expr| {
        if let Some(column) = expr.downcast_ref::<phys_expr::Column>()
            && column == column_old
        {
            return Ok(Transformed::yes(Arc::new(column_new.clone())));
        }

        Ok(Transformed::no(expr))
    })
    .data()
}

fn reverse_operator(op: Operator) -> Result<Operator> {
    op.swap().ok_or_else(|| {
        internal_datafusion_err!(
            "Could not reverse operator {op} while building pruning predicate"
        )
    })
}

/// Given a column reference to `column`, returns a pruning
/// expression in terms of the min and max that will evaluate to true
/// if the column may contain values, and false if definitely does not
/// contain values
fn build_single_column_expr(
    column: &phys_expr::Column,
    schema: &Schema,
    required_columns: &mut RequiredColumns,
    is_not: bool, // if true, treat as !col
) -> Option<Arc<dyn PhysicalExpr>> {
    let field = schema.field_with_name(column.name()).ok()?;

    if *field.data_type() == DataType::Boolean {
        let col_ref = Arc::new(column.clone()) as _;

        let min = required_columns
            .min_column_expr(column, &col_ref, field)
            .ok()?;
        let max = required_columns
            .max_column_expr(column, &col_ref, field)
            .ok()?;

        // remember -- we want an expression that is:
        // TRUE: if there may be rows that match
        // FALSE: if there are no rows that match
        if is_not {
            // The only way we know a column couldn't match is if both the min and max are true
            // !(min && max)
            Some(Arc::new(phys_expr::NotExpr::new(Arc::new(
                phys_expr::BinaryExpr::new(min, Operator::And, max),
            ))))
        } else {
            // the only way we know a column couldn't match is if both the min and max are false
            // !(!min && !max) --> min || max
            Some(Arc::new(phys_expr::BinaryExpr::new(min, Operator::Or, max)))
        }
    } else {
        None
    }
}

/// Given an expression reference to `expr`, if `expr` is a column expression,
/// returns a pruning expression in terms of IsNull that will evaluate to true
/// if the column may contain null, and false if definitely does not
/// contain null.
/// If `with_not` is true, build a pruning expression for `col IS NOT NULL`: `col_count != col_null_count`
/// The pruning expression evaluates to true ONLY if the column definitely CONTAINS
/// at least one NULL value.  In this case we can know that `IS NOT NULL` can not be true and
/// thus can prune the row group / value
fn build_is_null_column_expr(
    expr: &Arc<dyn PhysicalExpr>,
    schema: &Schema,
    required_columns: &mut RequiredColumns,
    with_not: bool,
) -> Option<Arc<dyn PhysicalExpr>> {
    if let Some(col) = expr.downcast_ref::<phys_expr::Column>() {
        let field = schema.field_with_name(col.name()).ok()?;

        let null_count_field = &Field::new(field.name(), DataType::UInt64, true);
        if with_not {
            if let Ok(row_count_expr) =
                required_columns.row_count_column_expr(col, expr, null_count_field)
            {
                required_columns
                    .null_count_column_expr(col, expr, null_count_field)
                    .map(|null_count_column_expr| {
                        // IsNotNull(column) => null_count != row_count
                        Arc::new(phys_expr::BinaryExpr::new(
                            null_count_column_expr,
                            Operator::NotEq,
                            row_count_expr,
                        )) as _
                    })
                    .ok()
            } else {
                None
            }
        } else {
            required_columns
                .null_count_column_expr(col, expr, null_count_field)
                .map(|null_count_column_expr| {
                    // IsNull(column) => null_count > 0
                    Arc::new(phys_expr::BinaryExpr::new(
                        null_count_column_expr,
                        Operator::Gt,
                        Arc::new(phys_expr::Literal::new(ScalarValue::UInt64(Some(0)))),
                    )) as _
                })
                .ok()
        }
    } else {
        None
    }
}

/// The maximum number of entries in an `InList` that might be rewritten into
/// an OR chain
const MAX_LIST_VALUE_SIZE_REWRITE: usize = 20;

/// Rewrite a predicate expression in terms of statistics (min/max/null_counts)
/// for use as a [`PruningPredicate`].
pub struct PredicateRewriter {
    unhandled_hook: Arc<dyn UnhandledPredicateHook>,
}

impl Default for PredicateRewriter {
    fn default() -> Self {
        Self {
            unhandled_hook: Arc::new(ConstantUnhandledPredicateHook::default()),
        }
    }
}

impl PredicateRewriter {
    /// Create a new `PredicateRewriter`
    pub fn new() -> Self {
        Self::default()
    }

    /// Set the unhandled hook to be used when a predicate can not be rewritten
    pub fn with_unhandled_hook(
        self,
        unhandled_hook: Arc<dyn UnhandledPredicateHook>,
    ) -> Self {
        Self { unhandled_hook }
    }

    /// Translate logical filter expression into pruning predicate
    /// expression that will evaluate to FALSE if it can be determined no
    /// rows between the min/max values could pass the predicates.
    ///
    /// Any predicates that can not be translated will be passed to `unhandled_hook`.
    ///
    /// Returns the pruning predicate as an [`PhysicalExpr`]
    ///
    /// Notice: Does not handle [`phys_expr::InListExpr`] greater than 20, which will fall back to calling `unhandled_hook`
    pub fn rewrite_predicate_to_statistics_predicate(
        &self,
        expr: &Arc<dyn PhysicalExpr>,
        schema: &Schema,
    ) -> Arc<dyn PhysicalExpr> {
        let mut required_columns = RequiredColumns::new();
        build_predicate_expression(
            expr,
            &Arc::new(schema.clone()),
            &mut required_columns,
            &self.unhandled_hook,
        )
    }
}

/// Translate logical filter expression into pruning predicate
/// expression that will evaluate to FALSE if it can be determined no
/// rows between the min/max values could pass the predicates.
///
/// Any predicates that can not be translated will be passed to `unhandled_hook`.
///
/// Returns the pruning predicate as an [`PhysicalExpr`]
///
/// Notice: Does not handle [`phys_expr::InListExpr`] greater than 20, which will fall back to calling `unhandled_hook`
fn build_predicate_expression(
    expr: &Arc<dyn PhysicalExpr>,
    schema: &SchemaRef,
    required_columns: &mut RequiredColumns,
    unhandled_hook: &Arc<dyn UnhandledPredicateHook>,
) -> Arc<dyn PhysicalExpr> {
    if is_always_false(expr) {
        // Shouldn't return `unhandled_hook.handle(expr)`
        // Because it will transfer false to true.
        return Arc::clone(expr);
    }
    // predicate expression can only be a binary expression
    if let Some(is_null) = expr.downcast_ref::<phys_expr::IsNullExpr>() {
        return build_is_null_column_expr(is_null.arg(), schema, required_columns, false)
            .unwrap_or_else(|| unhandled_hook.handle(expr));
    }
    if let Some(is_not_null) = expr.downcast_ref::<phys_expr::IsNotNullExpr>() {
        return build_is_null_column_expr(
            is_not_null.arg(),
            schema,
            required_columns,
            true,
        )
        .unwrap_or_else(|| unhandled_hook.handle(expr));
    }
    if let Some(col) = expr.downcast_ref::<phys_expr::Column>() {
        return build_single_column_expr(col, schema, required_columns, false)
            .unwrap_or_else(|| unhandled_hook.handle(expr));
    }
    if let Some(not) = expr.downcast_ref::<phys_expr::NotExpr>() {
        // match !col (don't do so recursively)
        if let Some(col) = not.arg().downcast_ref::<phys_expr::Column>() {
            return build_single_column_expr(col, schema, required_columns, true)
                .unwrap_or_else(|| unhandled_hook.handle(expr));
        } else {
            return unhandled_hook.handle(expr);
        }
    }
    if let Some(in_list) = expr.downcast_ref::<phys_expr::InListExpr>() {
        if !in_list.list().is_empty()
            && in_list.list().len() <= MAX_LIST_VALUE_SIZE_REWRITE
        {
            let eq_op = if in_list.negated() {
                Operator::NotEq
            } else {
                Operator::Eq
            };
            let re_op = if in_list.negated() {
                Operator::And
            } else {
                Operator::Or
            };
            let change_expr = in_list
                .list()
                .iter()
                .map(|e| {
                    Arc::new(phys_expr::BinaryExpr::new(
                        Arc::clone(in_list.expr()),
                        eq_op,
                        Arc::clone(e),
                    )) as _
                })
                .reduce(|a, b| Arc::new(phys_expr::BinaryExpr::new(a, re_op, b)) as _)
                .unwrap();
            return build_predicate_expression(
                &change_expr,
                schema,
                required_columns,
                unhandled_hook,
            );
        } else {
            return unhandled_hook.handle(expr);
        }
    }

    let (left, op, right) = {
        if let Some(bin_expr) = expr.downcast_ref::<phys_expr::BinaryExpr>() {
            (
                Arc::clone(bin_expr.left()),
                *bin_expr.op(),
                Arc::clone(bin_expr.right()),
            )
        } else if let Some(like_expr) = expr.downcast_ref::<phys_expr::LikeExpr>() {
            if like_expr.case_insensitive() {
                return unhandled_hook.handle(expr);
            }
            let op = match (like_expr.negated(), like_expr.case_insensitive()) {
                (false, false) => Operator::LikeMatch,
                (true, false) => Operator::NotLikeMatch,
                (false, true) => Operator::ILikeMatch,
                (true, true) => Operator::NotILikeMatch,
            };
            (
                Arc::clone(like_expr.expr()),
                op,
                Arc::clone(like_expr.pattern()),
            )
        } else {
            return unhandled_hook.handle(expr);
        }
    };

    if op == Operator::And || op == Operator::Or {
        let left_expr =
            build_predicate_expression(&left, schema, required_columns, unhandled_hook);
        let right_expr =
            build_predicate_expression(&right, schema, required_columns, unhandled_hook);
        // simplify boolean expression if applicable
        let expr = match (&left_expr, op, &right_expr) {
            (left, Operator::And, right)
                if is_always_false(left) || is_always_false(right) =>
            {
                Arc::new(phys_expr::Literal::new(ScalarValue::Boolean(Some(false))))
            }
            (left, Operator::And, _) if is_always_true(left) => right_expr,
            (_, Operator::And, right) if is_always_true(right) => left_expr,
            (left, Operator::Or, right)
                if is_always_true(left) || is_always_true(right) =>
            {
                Arc::new(phys_expr::Literal::new(ScalarValue::Boolean(Some(true))))
            }
            (left, Operator::Or, _) if is_always_false(left) => right_expr,
            (_, Operator::Or, right) if is_always_false(right) => left_expr,

            _ => Arc::new(phys_expr::BinaryExpr::new(left_expr, op, right_expr)),
        };
        return expr;
    }

    let left_columns = ColumnReferenceCount::from_expression(&left);
    let right_columns = ColumnReferenceCount::from_expression(&right);
    let expr_builder = PruningExpressionBuilder::try_new(
        &left,
        &right,
        left_columns,
        right_columns,
        op,
        schema,
        required_columns,
    );
    let mut expr_builder = match expr_builder {
        Ok(builder) => builder,
        // allow partial failure in predicate expression generation
        // this can still produce a useful predicate when multiple conditions are joined using AND
        Err(e) => {
            debug!("Error building pruning expression: {e}");
            return unhandled_hook.handle(expr);
        }
    };

    build_statistics_expr(&mut expr_builder)
        .unwrap_or_else(|_| unhandled_hook.handle(expr))
}

/// Count of distinct column references in an expression.
/// This is the same as [`collect_columns`] but optimized to stop counting
/// once more than one distinct column is found.
///
/// For example, in expression `col1 + col2`, the count is `Many`.
/// In expression `col1 + 5`, the count is `One`.
/// In expression `5 + 10`, the count is `Zero`.
///
/// [`collect_columns`]: datafusion_physical_expr::utils::collect_columns
#[derive(Debug, PartialEq, Eq)]
enum ColumnReferenceCount {
    /// no column references
    Zero,
    /// Only one column reference
    One(phys_expr::Column),
    /// More than one column reference
    Many,
}

impl ColumnReferenceCount {
    /// Count the number of distinct column references in an expression
    fn from_expression(expr: &Arc<dyn PhysicalExpr>) -> Self {
        let mut seen = HashSet::<phys_expr::Column>::new();
        expr.apply(|expr| {
            if let Some(column) = expr.downcast_ref::<phys_expr::Column>() {
                seen.insert(column.clone());
                if seen.len() > 1 {
                    return Ok(TreeNodeRecursion::Stop);
                }
            }
            Ok(TreeNodeRecursion::Continue)
        })
        // pre_visit always returns OK, so this will always too
        .expect("no way to return error during recursion");
        match seen.len() {
            0 => ColumnReferenceCount::Zero,
            1 => ColumnReferenceCount::One(
                seen.into_iter().next().expect("just checked len==1"),
            ),
            _ => ColumnReferenceCount::Many,
        }
    }
}

fn build_statistics_expr(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    let statistics_expr: Arc<dyn PhysicalExpr> = match expr_builder.op() {
        Operator::NotEq => build_ne_statistics_expr(expr_builder)?,
        Operator::Eq => {
            // column = literal => (min, max) = literal => min <= literal && literal <= max
            // (column / 2) = 4 => (column_min / 2) <= 4 && 4 <= (column_max / 2)
            build_eq_statistics_expr(expr_builder)?
        }
        Operator::IsDistinctFrom => return build_is_distinct_from(expr_builder),
        Operator::IsNotDistinctFrom => return build_is_not_distinct_from(expr_builder),
        Operator::NotLikeMatch => build_not_like_match(expr_builder)?,
        Operator::LikeMatch => build_like_match(expr_builder).ok_or_else(|| {
            plan_datafusion_err!(
                "LIKE expression with wildcard at the beginning is not supported"
            )
        })?,
        Operator::Gt => {
            // column > literal => (min, max) > literal => max > literal
            Arc::new(phys_expr::BinaryExpr::new(
                expr_builder.max_column_expr()?,
                Operator::Gt,
                Arc::clone(expr_builder.scalar_expr()),
            ))
        }
        Operator::GtEq => {
            // column >= literal => (min, max) >= literal => max >= literal
            Arc::new(phys_expr::BinaryExpr::new(
                expr_builder.max_column_expr()?,
                Operator::GtEq,
                Arc::clone(expr_builder.scalar_expr()),
            ))
        }
        Operator::Lt => {
            // column < literal => (min, max) < literal => min < literal
            Arc::new(phys_expr::BinaryExpr::new(
                expr_builder.min_column_expr()?,
                Operator::Lt,
                Arc::clone(expr_builder.scalar_expr()),
            ))
        }
        Operator::LtEq => {
            // column <= literal => (min, max) <= literal => min <= literal
            Arc::new(phys_expr::BinaryExpr::new(
                expr_builder.min_column_expr()?,
                Operator::LtEq,
                Arc::clone(expr_builder.scalar_expr()),
            ))
        }
        // other expressions are not supported
        _ => {
            return plan_err!(
                "expressions other than (neq, eq, gt, gteq, lt, lteq) are not supported"
            );
        }
    };
    let statistics_expr = wrap_null_count_check_expr(statistics_expr, expr_builder)?;
    Ok(statistics_expr)
}

fn binary_expr(
    left: Arc<dyn PhysicalExpr>,
    op: Operator,
    right: Arc<dyn PhysicalExpr>,
) -> Arc<dyn PhysicalExpr> {
    Arc::new(phys_expr::BinaryExpr::new(left, op, right))
}

fn and_expr(
    left: Arc<dyn PhysicalExpr>,
    right: Arc<dyn PhysicalExpr>,
) -> Arc<dyn PhysicalExpr> {
    binary_expr(left, Operator::And, right)
}

fn or_expr(
    left: Arc<dyn PhysicalExpr>,
    right: Arc<dyn PhysicalExpr>,
) -> Arc<dyn PhysicalExpr> {
    binary_expr(left, Operator::Or, right)
}

fn build_eq_statistics_expr(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    let min_column_expr = expr_builder.min_column_expr()?;
    let max_column_expr = expr_builder.max_column_expr()?;
    Ok(and_expr(
        binary_expr(
            min_column_expr,
            Operator::LtEq,
            Arc::clone(expr_builder.scalar_expr()),
        ),
        binary_expr(
            Arc::clone(expr_builder.scalar_expr()),
            Operator::LtEq,
            max_column_expr,
        ),
    ))
}

fn build_ne_statistics_expr(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    let min_column_expr = expr_builder.min_column_expr()?;
    let max_column_expr = expr_builder.max_column_expr()?;
    Ok(or_expr(
        binary_expr(
            min_column_expr,
            Operator::NotEq,
            Arc::clone(expr_builder.scalar_expr()),
        ),
        binary_expr(
            Arc::clone(expr_builder.scalar_expr()),
            Operator::NotEq,
            max_column_expr,
        ),
    ))
}

fn column_has_nulls_expr(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    Ok(binary_expr(
        expr_builder.null_count_column_expr()?,
        Operator::Gt,
        Arc::new(phys_expr::Literal::new(ScalarValue::UInt64(Some(0)))),
    ))
}

fn column_has_non_nulls_expr(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    Ok(binary_expr(
        expr_builder.null_count_column_expr()?,
        Operator::NotEq,
        expr_builder.row_count_column_expr()?,
    ))
}

fn build_is_distinct_from(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    let scalar_expr = Arc::clone(expr_builder.scalar_expr());

    Ok(or_expr(
        and_expr(
            Arc::new(phys_expr::IsNullExpr::new(Arc::clone(&scalar_expr))),
            column_has_non_nulls_expr(expr_builder)?,
        ),
        and_expr(
            Arc::new(phys_expr::IsNotNullExpr::new(scalar_expr)),
            or_expr(
                column_has_nulls_expr(expr_builder)?,
                build_ne_statistics_expr(expr_builder)?,
            ),
        ),
    ))
}

fn build_is_not_distinct_from(
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    let scalar_expr = Arc::clone(expr_builder.scalar_expr());

    Ok(or_expr(
        and_expr(
            Arc::new(phys_expr::IsNullExpr::new(Arc::clone(&scalar_expr))),
            column_has_nulls_expr(expr_builder)?,
        ),
        and_expr(
            Arc::new(phys_expr::IsNotNullExpr::new(scalar_expr)),
            and_expr(
                column_has_non_nulls_expr(expr_builder)?,
                build_eq_statistics_expr(expr_builder)?,
            ),
        ),
    ))
}

/// returns the string literal of the scalar value if it is a string
fn unpack_string(s: &ScalarValue) -> Option<&str> {
    s.try_as_str().flatten()
}

fn extract_string_literal(expr: &Arc<dyn PhysicalExpr>) -> Option<&str> {
    if let Some(lit) = expr.downcast_ref::<phys_expr::Literal>() {
        let s = unpack_string(lit.value())?;
        return Some(s);
    }
    None
}

/// Convert `column LIKE literal` where P is a constant prefix of the literal
/// to a range check on the column: `P <= column && column < P'`, where P' is the
/// lowest string after all P* strings.
fn build_like_match(
    expr_builder: &mut PruningExpressionBuilder,
) -> Option<Arc<dyn PhysicalExpr>> {
    // column LIKE literal => (min, max) LIKE literal split at unescaped % => min <= split literal && split literal <= max
    // column LIKE 'foo%' => min <= 'foo' && 'foo' <= max
    // column LIKE 'foo\_%' => min <= 'foo_' && 'foo_' <= max (the _ is escaped)
    // column LIKE 'foo\%%' => min <= 'foo%' && 'foo%' <= max (the % is escaped)
    // column LIKE '%foo' => min <= '' && '' <= max => true
    // column LIKE '%foo%' => min <= '' && '' <= max => true
    // column LIKE 'foo' => min <= 'foo' && 'foo' <= max

    // TODO Handle ILIKE perhaps by making the min lowercase and max uppercase
    //  this may involve building the physical expressions that call lower() and upper()
    let min_column_expr = expr_builder.min_column_expr().ok()?;
    let max_column_expr = expr_builder.max_column_expr().ok()?;
    let scalar_expr = expr_builder.scalar_expr();
    // check that the scalar is a string literal
    let s = extract_string_literal(scalar_expr)?;
    // ANSI SQL specifies two wildcards: % and _. % matches zero or more characters, _ matches exactly one character.
    let (decoded_prefix, rest) = split_constant_prefix(s);
    let has_wildcard = !rest.is_empty();
    if has_wildcard && decoded_prefix.is_empty() {
        // there's no filtering we could possibly do, return None and have this be handled by the unhandled hook
        return None;
    }
    let (lower_bound, upper_bound) = if has_wildcard {
        let incremented_prefix = increment_utf8(&decoded_prefix)?;
        let lower_bound_lit = Arc::new(phys_expr::Literal::new(ScalarValue::Utf8(Some(
            decoded_prefix,
        ))));
        let upper_bound_lit = Arc::new(phys_expr::Literal::new(ScalarValue::Utf8(Some(
            incremented_prefix,
        ))));
        (lower_bound_lit, upper_bound_lit)
    } else {
        // the like expression is a literal and can be converted into a comparison
        let bound = Arc::new(phys_expr::Literal::new(ScalarValue::Utf8(Some(
            decoded_prefix,
        ))));
        (Arc::clone(&bound), bound)
    };
    let lower_bound_expr = Arc::new(phys_expr::BinaryExpr::new(
        lower_bound,
        Operator::LtEq,
        Arc::clone(&max_column_expr),
    ));
    let upper_bound_expr = Arc::new(phys_expr::BinaryExpr::new(
        Arc::clone(&min_column_expr),
        Operator::LtEq,
        upper_bound,
    ));
    let combined = Arc::new(phys_expr::BinaryExpr::new(
        upper_bound_expr,
        Operator::And,
        lower_bound_expr,
    ));
    Some(combined)
}

// For predicate `col NOT LIKE 'const_prefix%'`, we rewrite it as `(col_min NOT LIKE 'const_prefix%' OR col_max NOT LIKE 'const_prefix%')`.
//
// The intuition is that if both `col_min` and `col_max` begin with `const_prefix` that means
// **all** data in this row group begins with `const_prefix` as well (and therefore the predicate
// looking for rows that don't begin with `const_prefix` can never be true)
fn build_not_like_match(
    expr_builder: &mut PruningExpressionBuilder<'_>,
) -> Result<Arc<dyn PhysicalExpr>> {
    // col NOT LIKE 'const_prefix%' -> !(col_min LIKE 'const_prefix%' && col_max LIKE 'const_prefix%') -> (col_min NOT LIKE 'const_prefix%' || col_max NOT LIKE 'const_prefix%')

    let min_column_expr = expr_builder.min_column_expr()?;
    let max_column_expr = expr_builder.max_column_expr()?;

    let scalar_expr = expr_builder.scalar_expr();

    let pattern = extract_string_literal(scalar_expr).ok_or_else(|| {
        plan_datafusion_err!("cannot extract literal from NOT LIKE expression")
    })?;

    let (const_prefix, remaining) = split_constant_prefix(pattern);
    if const_prefix.is_empty() || remaining != "%" {
        // we can not handle `%` at the beginning or in the middle of the pattern
        // Example: For pattern "foo%bar", the row group might include values like
        // ["foobar", "food", "foodbar"], making it unsafe to prune.
        // Even if the min/max values in the group (e.g., "foobar" and "foodbar")
        // match the pattern, intermediate values like "food" may not
        // match the full pattern "foo%bar", making pruning unsafe.
        // (truncate foo%bar to foo% have same problem)

        // we can not handle pattern containing `_`
        // Example: For pattern "foo_", row groups might contain ["fooa", "fooaa", "foob"],
        // which means not every row is guaranteed to match the pattern.
        return Err(plan_datafusion_err!(
            "NOT LIKE expressions only support constant_prefix+wildcard`%`"
        ));
    }

    let min_col_not_like_epxr = Arc::new(phys_expr::LikeExpr::new(
        true,
        false,
        Arc::clone(&min_column_expr),
        Arc::clone(scalar_expr),
    ));

    let max_col_not_like_expr = Arc::new(phys_expr::LikeExpr::new(
        true,
        false,
        Arc::clone(&max_column_expr),
        Arc::clone(scalar_expr),
    ));

    Ok(Arc::new(phys_expr::BinaryExpr::new(
        min_col_not_like_epxr,
        Operator::Or,
        max_col_not_like_expr,
    )))
}

/// Returns unescaped constant prefix of a LIKE pattern (possibly empty) and the remaining pattern (possibly empty)
fn split_constant_prefix(pattern: &str) -> (String, &str) {
    let mut prefix = String::with_capacity(pattern.len());
    let mut iter = pattern.char_indices();
    while let Some((idx, c)) = iter.next() {
        match c {
            '%' | '_' => return (prefix, &pattern[idx..]),
            '\\' => match iter.next() {
                Some((_, escaped)) => prefix.push(escaped),
                None => prefix.push('\\'),
            },
            _ => prefix.push(c),
        }
    }
    (prefix, "")
}

/// Increment a UTF8 string by one, returning `None` if it can't be incremented.
/// This makes it so that the returned string will always compare greater than the input string
/// or any other string with the same prefix.
/// This is necessary since the statistics may have been truncated: if we have a min statistic
/// of "fo" that may have originally been "foz" or anything else with the prefix "fo".
/// E.g. `increment_utf8("foo") >= "foo"` and `increment_utf8("foo") >= "fooz"`
/// In this example `increment_utf8("foo") == "fop"
fn increment_utf8(data: &str) -> Option<String> {
    // Helper function to check if a character is valid to use
    fn is_valid_unicode(c: char) -> bool {
        let cp = c as u32;

        // Filter out non-characters (https://www.unicode.org/versions/corrigendum9.html)
        if [0xFFFE, 0xFFFF].contains(&cp) || (0xFDD0..=0xFDEF).contains(&cp) {
            return false;
        }

        // Filter out private use area
        if cp >= 0x110000 {
            return false;
        }

        true
    }

    // Convert string to vector of code points
    let mut code_points: Vec<char> = data.chars().collect();

    // Work backwards through code points
    for idx in (0..code_points.len()).rev() {
        let original = code_points[idx] as u32;

        // Try incrementing the code point
        if let Some(next_char) = char::from_u32(original + 1)
            && is_valid_unicode(next_char)
        {
            code_points[idx] = next_char;
            // truncate the string to the current index
            code_points.truncate(idx + 1);
            return Some(code_points.into_iter().collect());
        }
    }

    None
}

/// Wrap the statistics expression in a check that skips the expression if the column is all nulls.
///
/// This is important not only as an optimization but also because statistics may not be
/// accurate for columns that are all nulls.
/// For example, for an `int` column `x` with all nulls, the min/max/null_count statistics
/// might be set to 0 and evaluating `x = 0` would incorrectly include the column.
///
/// For example:
///
/// `x_min <= 10 AND 10 <= x_max`
///
/// will become
///
/// ```sql
/// x_null_count != x_row_count AND (x_min <= 10 AND 10 <= x_max)
/// ````
///
/// If the column is known to be all nulls, then the expression
/// `x_null_count = x_row_count` will be true, which will cause the
/// boolean expression to return false. Therefore, prune out the container.
fn wrap_null_count_check_expr(
    statistics_expr: Arc<dyn PhysicalExpr>,
    expr_builder: &mut PruningExpressionBuilder,
) -> Result<Arc<dyn PhysicalExpr>> {
    // (x_null_count != x_row_count) AND (<statistics_expr>)
    Ok(and_expr(
        column_has_non_nulls_expr(expr_builder)?,
        statistics_expr,
    ))
}

#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub(crate) enum StatisticsType {
    Min,
    Max,
    NullCount,
    RowCount,
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::ops::{Not, Rem};

    use super::*;
    use datafusion_common::test_util::batches_to_string;
    use datafusion_expr::{and, col, lit, or};
    use datafusion_physical_expr::utils::collect_columns;
    use insta::assert_snapshot;

    use arrow::array::Decimal128Array;
    use arrow::{
        array::{BinaryArray, Int32Array, Int64Array, StringArray, UInt64Array},
        datatypes::TimeUnit,
    };
    use datafusion_expr::expr::InList;
    use datafusion_expr::{BinaryExpr, Expr, cast, is_null, try_cast};
    use datafusion_functions_nested::expr_fn::{array_has, make_array};
    use datafusion_physical_expr::expressions::{
        self as phys_expr, DynamicFilterPhysicalExpr,
    };
    use datafusion_physical_expr::planner::logical2physical;
    use itertools::Itertools;

    #[derive(Debug, Default)]
    /// Mock statistic provider for tests
    ///
    /// Each row represents the statistics for a "container" (which
    /// might represent an entire parquet file, or directory of files,
    /// or some other collection of data for which we had statistics)
    ///
    /// Note All `ArrayRefs` must be the same size.
    struct ContainerStats {
        min: Option<ArrayRef>,
        max: Option<ArrayRef>,
        /// Optional values
        null_counts: Option<ArrayRef>,
        row_counts: Option<ArrayRef>,
        /// Optional known values (e.g. mimic a bloom filter)
        /// (value, contained)
        /// If present, all BooleanArrays must be the same size as min/max
        contained: Vec<(HashSet<ScalarValue>, BooleanArray)>,
    }

    impl ContainerStats {
        fn new() -> Self {
            Default::default()
        }
        fn new_decimal128(
            min: impl IntoIterator<Item = Option<i128>>,
            max: impl IntoIterator<Item = Option<i128>>,
            precision: u8,
            scale: i8,
        ) -> Self {
            Self::new()
                .with_min(Arc::new(
                    min.into_iter()
                        .collect::<Decimal128Array>()
                        .with_precision_and_scale(precision, scale)
                        .unwrap(),
                ))
                .with_max(Arc::new(
                    max.into_iter()
                        .collect::<Decimal128Array>()
                        .with_precision_and_scale(precision, scale)
                        .unwrap(),
                ))
        }

        fn new_i64(
            min: impl IntoIterator<Item = Option<i64>>,
            max: impl IntoIterator<Item = Option<i64>>,
        ) -> Self {
            Self::new()
                .with_min(Arc::new(min.into_iter().collect::<Int64Array>()))
                .with_max(Arc::new(max.into_iter().collect::<Int64Array>()))
        }

        fn new_i32(
            min: impl IntoIterator<Item = Option<i32>>,
            max: impl IntoIterator<Item = Option<i32>>,
        ) -> Self {
            Self::new()
                .with_min(Arc::new(min.into_iter().collect::<Int32Array>()))
                .with_max(Arc::new(max.into_iter().collect::<Int32Array>()))
        }

        fn new_utf8<'a>(
            min: impl IntoIterator<Item = Option<&'a str>>,
            max: impl IntoIterator<Item = Option<&'a str>>,
        ) -> Self {
            Self::new()
                .with_min(Arc::new(min.into_iter().collect::<StringArray>()))
                .with_max(Arc::new(max.into_iter().collect::<StringArray>()))
        }

        fn new_bool(
            min: impl IntoIterator<Item = Option<bool>>,
            max: impl IntoIterator<Item = Option<bool>>,
        ) -> Self {
            Self::new()
                .with_min(Arc::new(min.into_iter().collect::<BooleanArray>()))
                .with_max(Arc::new(max.into_iter().collect::<BooleanArray>()))
        }

        fn min(&self) -> Option<ArrayRef> {
            self.min.clone()
        }

        fn max(&self) -> Option<ArrayRef> {
            self.max.clone()
        }

        fn null_counts(&self) -> Option<ArrayRef> {
            self.null_counts.clone()
        }

        fn row_counts(&self) -> Option<ArrayRef> {
            self.row_counts.clone()
        }

        /// return an iterator over all arrays in this statistics
        fn arrays(&self) -> Vec<ArrayRef> {
            let contained_arrays = self
                .contained
                .iter()
                .map(|(_values, contained)| Arc::new(contained.clone()) as ArrayRef);

            [
                self.min.as_ref().cloned(),
                self.max.as_ref().cloned(),
                self.null_counts.as_ref().cloned(),
                self.row_counts.as_ref().cloned(),
            ]
            .into_iter()
            .flatten()
            .chain(contained_arrays)
            .collect()
        }

        /// Returns the number of containers represented by this statistics This
        /// picks the length of the first array as all arrays must have the same
        /// length (which is verified by `assert_invariants`).
        fn len(&self) -> usize {
            // pick the first non zero length
            self.arrays().iter().map(|a| a.len()).next().unwrap_or(0)
        }

        /// Ensure that the lengths of all arrays are consistent
        fn assert_invariants(&self) {
            let mut prev_len = None;

            for len in self.arrays().iter().map(|a| a.len()) {
                // Get a length, if we don't already have one
                match prev_len {
                    None => {
                        prev_len = Some(len);
                    }
                    Some(prev_len) => {
                        assert_eq!(prev_len, len);
                    }
                }
            }
        }

        /// Add min values
        fn with_min(mut self, min: ArrayRef) -> Self {
            self.min = Some(min);
            self
        }

        /// Add max values
        fn with_max(mut self, max: ArrayRef) -> Self {
            self.max = Some(max);
            self
        }

        /// Add null counts. There must be the same number of null counts as
        /// there are containers
        fn with_null_counts(
            mut self,
            counts: impl IntoIterator<Item = Option<u64>>,
        ) -> Self {
            let null_counts: ArrayRef =
                Arc::new(counts.into_iter().collect::<UInt64Array>());

            self.assert_invariants();
            self.null_counts = Some(null_counts);
            self
        }

        /// Add row counts. There must be the same number of row counts as
        /// there are containers
        fn with_row_counts(
            mut self,
            counts: impl IntoIterator<Item = Option<u64>>,
        ) -> Self {
            let row_counts: ArrayRef =
                Arc::new(counts.into_iter().collect::<UInt64Array>());

            self.assert_invariants();
            self.row_counts = Some(row_counts);
            self
        }

        /// Add contained information.
        #[allow(clippy::allow_attributes, clippy::mutable_key_type)] // ScalarValue has interior mutability but is intentionally used as hash key
        pub fn with_contained(
            mut self,
            values: impl IntoIterator<Item = ScalarValue>,
            contained: impl IntoIterator<Item = Option<bool>>,
        ) -> Self {
            let contained: BooleanArray = contained.into_iter().collect();
            let values: HashSet<_> = values.into_iter().collect();

            self.contained.push((values, contained));
            self.assert_invariants();
            self
        }

        /// get any contained information for the specified values
        #[allow(clippy::allow_attributes, clippy::mutable_key_type)] // ScalarValue has interior mutability but is intentionally used as hash key
        fn contained(&self, find_values: &HashSet<ScalarValue>) -> Option<BooleanArray> {
            // find the one with the matching values
            self.contained
                .iter()
                .find(|(values, _contained)| values == find_values)
                .map(|(_values, contained)| contained.clone())
        }
    }

    #[derive(Debug, Default)]
    struct TestStatistics {
        // key: column name
        stats: HashMap<Column, ContainerStats>,
    }

    impl TestStatistics {
        fn new() -> Self {
            Self::default()
        }

        fn with(
            mut self,
            name: impl Into<String>,
            container_stats: ContainerStats,
        ) -> Self {
            let col = Column::from_name(name.into());
            self.stats.insert(col, container_stats);
            self
        }

        /// Add null counts for the specified column.
        /// There must be the same number of null counts as
        /// there are containers
        fn with_null_counts(
            mut self,
            name: impl Into<String>,
            counts: impl IntoIterator<Item = Option<u64>>,
        ) -> Self {
            let col = Column::from_name(name.into());

            // take stats out and update them
            let container_stats = self
                .stats
                .remove(&col)
                .unwrap_or_default()
                .with_null_counts(counts);

            // put stats back in
            self.stats.insert(col, container_stats);
            self
        }

        /// Add row counts for the specified column.
        /// There must be the same number of row counts as
        /// there are containers
        fn with_row_counts(
            mut self,
            name: impl Into<String>,
            counts: impl IntoIterator<Item = Option<u64>>,
        ) -> Self {
            let col = Column::from_name(name.into());

            // take stats out and update them
            let container_stats = self
                .stats
                .remove(&col)
                .unwrap_or_default()
                .with_row_counts(counts);

            // put stats back in
            self.stats.insert(col, container_stats);
            self
        }

        /// Add contained information for the specified column.
        fn with_contained(
            mut self,
            name: impl Into<String>,
            values: impl IntoIterator<Item = ScalarValue>,
            contained: impl IntoIterator<Item = Option<bool>>,
        ) -> Self {
            let col = Column::from_name(name.into());

            // take stats out and update them
            let container_stats = self
                .stats
                .remove(&col)
                .unwrap_or_default()
                .with_contained(values, contained);

            // put stats back in
            self.stats.insert(col, container_stats);
            self
        }
    }

    impl PruningStatistics for TestStatistics {
        fn min_values(&self, column: &Column) -> Option<ArrayRef> {
            self.stats
                .get(column)
                .map(|container_stats| container_stats.min())
                .unwrap_or(None)
        }

        fn max_values(&self, column: &Column) -> Option<ArrayRef> {
            self.stats
                .get(column)
                .map(|container_stats| container_stats.max())
                .unwrap_or(None)
        }

        fn num_containers(&self) -> usize {
            self.stats
                .values()
                .next()
                .map(|container_stats| container_stats.len())
                .unwrap_or(0)
        }

        fn null_counts(&self, column: &Column) -> Option<ArrayRef> {
            self.stats
                .get(column)
                .map(|container_stats| container_stats.null_counts())
                .unwrap_or(None)
        }

        fn row_counts(&self) -> Option<ArrayRef> {
            self.stats
                .values()
                .find_map(|container_stats| container_stats.row_counts())
        }

        fn contained(
            &self,
            column: &Column,
            values: &HashSet<ScalarValue>,
        ) -> Option<BooleanArray> {
            self.stats
                .get(column)
                .and_then(|container_stats| container_stats.contained(values))
        }
    }

    /// Returns the specified min/max container values
    struct OneContainerStats {
        min_values: Option<ArrayRef>,
        max_values: Option<ArrayRef>,
        num_containers: usize,
    }

    impl PruningStatistics for OneContainerStats {
        fn min_values(&self, _column: &Column) -> Option<ArrayRef> {
            self.min_values.clone()
        }

        fn max_values(&self, _column: &Column) -> Option<ArrayRef> {
            self.max_values.clone()
        }

        fn num_containers(&self) -> usize {
            self.num_containers
        }

        fn null_counts(&self, _column: &Column) -> Option<ArrayRef> {
            None
        }

        fn row_counts(&self) -> Option<ArrayRef> {
            None
        }

        fn contained(
            &self,
            _column: &Column,
            _values: &HashSet<ScalarValue>,
        ) -> Option<BooleanArray> {
            None
        }
    }

    /// Row count should only be referenced once in the pruning expression, even if we need the row count
    /// for multiple columns.
    #[test]
    fn test_unique_row_count_field_and_column() {
        // c1 = 100 AND c2 = 200
        let schema: SchemaRef = Arc::new(Schema::new(vec![
            Field::new("c1", DataType::Int32, true),
            Field::new("c2", DataType::Int32, true),
        ]));
        let expr = col("c1").eq(lit(100)).and(col("c2").eq(lit(200)));
        let expr = logical2physical(&expr, &schema);
        let p = PruningPredicate::try_new(expr, Arc::clone(&schema)).unwrap();
        // note pruning expression refers to row_count twice
        assert_eq!(
            "c1_null_count@2 != row_count@3 AND c1_min@0 <= 100 AND 100 <= c1_max@1 AND c2_null_count@6 != row_count@3 AND c2_min@4 <= 200 AND 200 <= c2_max@5",
            p.predicate_expr.to_string()
        );

        // Fields in required schema should be unique, otherwise when creating batches
        // it will fail because of duplicate field names
        let mut fields = HashSet::new();
        for (_col, _ty, field) in p.required_columns().iter() {
            let was_new = fields.insert(field);
            if !was_new {
                panic!(
                    "Duplicate field in required schema: {field:?}. Previous fields:\n{fields:#?}"
                );
            }
        }
    }

    #[test]
    fn prune_all_rows_null_counts() {
        // if null_count = row_count then we should prune the container for i = 0
        // regardless of the statistics
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
        let statistics = TestStatistics::new().with(
            "i",
            ContainerStats::new_i32(
                vec![Some(0)], // min
                vec![Some(0)], // max
            )
            .with_null_counts(vec![Some(1)])
            .with_row_counts(vec![Some(1)]),
        );
        let expected_ret = &[false];
        prune_with_expr(col("i").eq(lit(0)), &schema, &statistics, expected_ret);

        // this should be true even if the container stats are missing
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
        let container_stats = ContainerStats {
            min: Some(Arc::new(Int32Array::from(vec![None]))),
            max: Some(Arc::new(Int32Array::from(vec![None]))),
            null_counts: Some(Arc::new(UInt64Array::from(vec![Some(1)]))),
            row_counts: Some(Arc::new(UInt64Array::from(vec![Some(1)]))),
            ..ContainerStats::default()
        };
        let statistics = TestStatistics::new().with("i", container_stats);
        let expected_ret = &[false];
        prune_with_expr(col("i").eq(lit(0)), &schema, &statistics, expected_ret);

        // If the null counts themselves are missing we should be able to fall back to the stats
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
        let container_stats = ContainerStats {
            min: Some(Arc::new(Int32Array::from(vec![Some(0)]))),
            max: Some(Arc::new(Int32Array::from(vec![Some(0)]))),
            null_counts: Some(Arc::new(UInt64Array::from(vec![None]))),
            row_counts: Some(Arc::new(UInt64Array::from(vec![Some(1)]))),
            ..ContainerStats::default()
        };
        let statistics = TestStatistics::new().with("i", container_stats);
        let expected_ret = &[true];
        prune_with_expr(col("i").eq(lit(0)), &schema, &statistics, expected_ret);
        let expected_ret = &[false];
        prune_with_expr(col("i").gt(lit(0)), &schema, &statistics, expected_ret);

        // Same for the row counts
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
        let container_stats = ContainerStats {
            min: Some(Arc::new(Int32Array::from(vec![Some(0)]))),
            max: Some(Arc::new(Int32Array::from(vec![Some(0)]))),
            null_counts: Some(Arc::new(UInt64Array::from(vec![Some(1)]))),
            row_counts: Some(Arc::new(UInt64Array::from(vec![None]))),
            ..ContainerStats::default()
        };
        let statistics = TestStatistics::new().with("i", container_stats);
        let expected_ret = &[true];
        prune_with_expr(col("i").eq(lit(0)), &schema, &statistics, expected_ret);
        let expected_ret = &[false];
        prune_with_expr(col("i").gt(lit(0)), &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_missing_statistics() {
        // If the min or max stats are missing we should not prune
        // (unless we know all rows are null, see `prune_all_rows_null_counts`)
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
        let container_stats = ContainerStats {
            min: Some(Arc::new(Int32Array::from(vec![None, Some(0)]))),
            max: Some(Arc::new(Int32Array::from(vec![Some(0), None]))),
            null_counts: Some(Arc::new(UInt64Array::from(vec![Some(0), Some(0)]))),
            row_counts: Some(Arc::new(UInt64Array::from(vec![Some(1), Some(1)]))),
            ..ContainerStats::default()
        };
        let statistics = TestStatistics::new().with("i", container_stats);
        let expected_ret = &[true, true];
        prune_with_expr(col("i").eq(lit(0)), &schema, &statistics, expected_ret);
        let expected_ret = &[false, true];
        prune_with_expr(col("i").gt(lit(0)), &schema, &statistics, expected_ret);
        let expected_ret = &[true, false];
        prune_with_expr(col("i").lt(lit(0)), &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_null_stats() {
        // if null_count = row_count then we should prune the container for i = 0
        // regardless of the statistics
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));

        let statistics = TestStatistics::new().with(
            "i",
            ContainerStats::new_i32(
                vec![Some(0)], // min
                vec![Some(0)], // max
            )
            .with_null_counts(vec![Some(1)])
            .with_row_counts(vec![Some(1)]),
        );

        let expected_ret = &[false];

        // i = 0
        prune_with_expr(col("i").eq(lit(0)), &schema, &statistics, expected_ret);
    }

    #[test]
    fn test_build_statistics_record_batch() {
        // Request a record batch with of s1_min, s2_max, s3_max, s3_min
        let required_columns = RequiredColumns::from(vec![
            // min of original column s1, named s1_min
            (
                phys_expr::Column::new("s1", 1),
                StatisticsType::Min,
                Field::new("s1_min", DataType::Int32, true),
            ),
            // max of original column s2, named s2_max
            (
                phys_expr::Column::new("s2", 2),
                StatisticsType::Max,
                Field::new("s2_max", DataType::Int32, true),
            ),
            // max of original column s3, named s3_max
            (
                phys_expr::Column::new("s3", 3),
                StatisticsType::Max,
                Field::new("s3_max", DataType::Utf8, true),
            ),
            // min of original column s3, named s3_min
            (
                phys_expr::Column::new("s3", 3),
                StatisticsType::Min,
                Field::new("s3_min", DataType::Utf8, true),
            ),
        ]);

        let statistics = TestStatistics::new()
            .with(
                "s1",
                ContainerStats::new_i32(
                    vec![None, None, Some(9), None],  // min
                    vec![Some(10), None, None, None], // max
                ),
            )
            .with(
                "s2",
                ContainerStats::new_i32(
                    vec![Some(2), None, None, None],  // min
                    vec![Some(20), None, None, None], // max
                ),
            )
            .with(
                "s3",
                ContainerStats::new_utf8(
                    vec![Some("a"), None, None, None],      // min
                    vec![Some("q"), None, Some("r"), None], // max
                ),
            );

        let batch =
            build_statistics_record_batch(&statistics, &required_columns).unwrap();
        assert_snapshot!(batches_to_string(&[batch]), @r"
        +--------+--------+--------+--------+
        | s1_min | s2_max | s3_max | s3_min |
        +--------+--------+--------+--------+
        |        | 20     | q      | a      |
        |        |        |        |        |
        | 9      |        | r      |        |
        |        |        |        |        |
        +--------+--------+--------+--------+
        ");
    }

    #[test]
    fn test_build_statistics_casting() {
        // Test requesting a Timestamp column, but getting statistics as Int64
        // which is what Parquet does

        // Request a record batch with of s1_min as a timestamp
        let required_columns = RequiredColumns::from(vec![(
            phys_expr::Column::new("s3", 3),
            StatisticsType::Min,
            Field::new(
                "s1_min",
                DataType::Timestamp(TimeUnit::Nanosecond, None),
                true,
            ),
        )]);

        // Note the statistics pass back i64 (not timestamp)
        let statistics = OneContainerStats {
            min_values: Some(Arc::new(Int64Array::from(vec![Some(10)]))),
            max_values: Some(Arc::new(Int64Array::from(vec![Some(20)]))),
            num_containers: 1,
        };

        let batch =
            build_statistics_record_batch(&statistics, &required_columns).unwrap();

        assert_snapshot!(batches_to_string(&[batch]), @r"
        +-------------------------------+
        | s1_min                        |
        +-------------------------------+
        | 1970-01-01T00:00:00.000000010 |
        +-------------------------------+
        ");
    }

    #[test]
    fn test_build_statistics_no_required_stats() {
        let required_columns = RequiredColumns::new();

        let statistics = OneContainerStats {
            min_values: Some(Arc::new(Int64Array::from(vec![Some(10)]))),
            max_values: Some(Arc::new(Int64Array::from(vec![Some(20)]))),
            num_containers: 1,
        };

        let batch =
            build_statistics_record_batch(&statistics, &required_columns).unwrap();
        assert_eq!(batch.num_rows(), 1); // had 1 container
    }

    #[test]
    fn test_build_statistics_inconsistent_types() {
        // Test requesting a Utf8 column when the stats return some other type

        // Request a record batch with of s1_min as a timestamp
        let required_columns = RequiredColumns::from(vec![(
            phys_expr::Column::new("s3", 3),
            StatisticsType::Min,
            Field::new("s1_min", DataType::Utf8, true),
        )]);

        // Note the statistics return an invalid UTF-8 sequence which will be converted to null
        let statistics = OneContainerStats {
            min_values: Some(Arc::new(BinaryArray::from(vec![&[255u8] as &[u8]]))),
            max_values: None,
            num_containers: 1,
        };

        let batch =
            build_statistics_record_batch(&statistics, &required_columns).unwrap();
        assert_snapshot!(batches_to_string(&[batch]), @r"
        +--------+
        | s1_min |
        +--------+
        |        |
        +--------+
        ");
    }

    #[test]
    fn test_build_statistics_inconsistent_length() {
        // return an inconsistent length to the actual statistics arrays
        let required_columns = RequiredColumns::from(vec![(
            phys_expr::Column::new("s1", 3),
            StatisticsType::Min,
            Field::new("s1_min", DataType::Int64, true),
        )]);

        // Note the statistics pass back i64 (not timestamp)
        let statistics = OneContainerStats {
            min_values: Some(Arc::new(Int64Array::from(vec![Some(10)]))),
            max_values: Some(Arc::new(Int64Array::from(vec![Some(20)]))),
            num_containers: 3,
        };

        let result =
            build_statistics_record_batch(&statistics, &required_columns).unwrap_err();
        assert!(
            result
                .to_string()
                .contains("mismatched statistics length. Expected 3, got 1"),
            "{}",
            result
        );
    }

    #[test]
    fn row_group_predicate_eq() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr =
            "c1_null_count@2 != row_count@3 AND c1_min@0 <= 1 AND 1 <= c1_max@1";

        // test column on the left
        let expr = col("c1").eq(lit(1));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(1).eq(col("c1"));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_not_eq() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr =
            "c1_null_count@2 != row_count@3 AND (c1_min@0 != 1 OR 1 != c1_max@1)";

        // test column on the left
        let expr = col("c1").not_eq(lit(1));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(1).not_eq(col("c1"));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_gt() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_max@0 > 1";

        // test column on the left
        let expr = col("c1").gt(lit(1));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(1).lt(col("c1"));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_gt_eq() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_max@0 >= 1";

        // test column on the left
        let expr = col("c1").gt_eq(lit(1));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);
        // test column on the right
        let expr = lit(1).lt_eq(col("c1"));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_lt() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_min@0 < 1";

        // test column on the left
        let expr = col("c1").lt(lit(1));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(1).gt(col("c1"));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_lt_eq() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_min@0 <= 1";

        // test column on the left
        let expr = col("c1").lt_eq(lit(1));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);
        // test column on the right
        let expr = lit(1).gt_eq(col("c1"));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_and() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
            Field::new("c3", DataType::Int32, false),
        ]);
        // test AND operator joining supported c1 < 1 expression and unsupported c2 > c3 expression
        let expr = col("c1").lt(lit(1)).and(col("c2").lt(col("c3")));
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_min@0 < 1";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_or() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);
        // test OR operator joining supported c1 < 1 expression and unsupported c2 % 2 = 0 expression
        let expr = col("c1").lt(lit(1)).or(col("c2").rem(lit(2)).eq(lit(0)));
        let expected_expr = "true";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_not() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "true";

        let expr = col("c1").not();
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_not_bool() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Boolean, false)]);
        let expected_expr = "NOT c1_min@0 AND c1_max@1";

        let expr = col("c1").not();
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_bool() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Boolean, false)]);
        let expected_expr = "c1_min@0 OR c1_max@1";

        let expr = col("c1");
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    /// Test that non-boolean literal expressions don't prune any containers and error gracefully by not pruning anything instead of e.g. panicking
    #[test]
    fn row_group_predicate_non_boolean() {
        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
        let statistics = TestStatistics::new()
            .with("c1", ContainerStats::new_i32(vec![Some(0)], vec![Some(10)]));
        let expected_ret = &[true];
        prune_with_expr(lit(1), &schema, &statistics, expected_ret);
    }

    // Test that literal-to-literal comparisons are correctly evaluated.
    // When both sides are constants, the expression should be evaluated directly
    // and if it's false, all containers should be pruned.
    #[test]
    fn row_group_predicate_literal_false() {
        // lit(1) = lit(2) is always false, so all containers should be pruned
        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
        let statistics = TestStatistics::new()
            .with("c1", ContainerStats::new_i32(vec![Some(0)], vec![Some(10)]));
        let expected_ret = &[false];
        prune_with_simplified_expr(lit(1).eq(lit(2)), &schema, &statistics, expected_ret);
    }

    /// Test nested/complex literal expression trees.
    /// This is an integration test that PhysicalExprSimplifier + PruningPredicate work together as expected.
    #[test]
    fn row_group_predicate_literal_true() {
        // lit(1) = lit(1) is always true, so no containers should be pruned
        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
        let statistics = TestStatistics::new()
            .with("c1", ContainerStats::new_i32(vec![Some(0)], vec![Some(10)]));
        let expected_ret = &[true];
        prune_with_simplified_expr(lit(1).eq(lit(1)), &schema, &statistics, expected_ret);
    }

    /// Test nested/complex literal expression trees.
    /// This is an integration test that PhysicalExprSimplifier + PruningPredicate work together as expected.
    #[test]
    fn row_group_predicate_literal_null() {
        // lit(1) = null is always null, so no containers should be pruned
        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
        let statistics = TestStatistics::new()
            .with("c1", ContainerStats::new_i32(vec![Some(0)], vec![Some(10)]));
        let expected_ret = &[true];
        prune_with_simplified_expr(
            lit(1).eq(lit(ScalarValue::Null)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    /// Test nested/complex literal expression trees.
    /// This is an integration test that PhysicalExprSimplifier + PruningPredicate work together as expected.
    #[test]
    fn row_group_predicate_complex_literals() {
        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
        let statistics = TestStatistics::new()
            .with("c1", ContainerStats::new_i32(vec![Some(0)], vec![Some(10)]));

        // (1 + 2) > 0 is always true
        prune_with_simplified_expr(
            (lit(1) + lit(2)).gt(lit(0)),
            &schema,
            &statistics,
            &[true],
        );

        // (1 + 2) < 0 is always false
        prune_with_simplified_expr(
            (lit(1) + lit(2)).lt(lit(0)),
            &schema,
            &statistics,
            &[false],
        );

        // Nested AND of literals: true AND false = false
        prune_with_simplified_expr(
            lit(true).and(lit(false)),
            &schema,
            &statistics,
            &[false],
        );

        // Nested OR of literals: true OR false = true
        prune_with_simplified_expr(
            lit(true).or(lit(false)),
            &schema,
            &statistics,
            &[true],
        );

        // Complex nested: (1 < 2) AND (3 > 1) = true AND true = true
        prune_with_simplified_expr(
            lit(1).lt(lit(2)).and(lit(3).gt(lit(1))),
            &schema,
            &statistics,
            &[true],
        );

        // Complex nested: (1 > 2) OR (3 < 1) = false OR false = false
        prune_with_simplified_expr(
            lit(1).gt(lit(2)).or(lit(3).lt(lit(1))),
            &schema,
            &statistics,
            &[false],
        );
    }

    /// Integration test demonstrating that a dynamic filter with replaced children as literals will be snapshotted, simplified and then pruned correctly.
    #[test]
    fn row_group_predicate_dynamic_filter_with_literals() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("c1", DataType::Int32, true),
            Field::new("part", DataType::Utf8, true),
        ]));
        let statistics = TestStatistics::new()
            // Note that we have no stats, pruning can only happen via partition value pruning from the dynamic filter
            .with_row_counts("c1", vec![Some(10)]);
        let dynamic_filter_expr = col("c1").gt(lit(5)).and(col("part").eq(lit("B")));
        let phys_expr = logical2physical(&dynamic_filter_expr, &schema);
        let children = collect_columns(&phys_expr)
            .iter()
            .map(|c| Arc::new(c.clone()) as Arc<dyn PhysicalExpr>)
            .collect_vec();
        let dynamic_phys_expr =
            Arc::new(DynamicFilterPhysicalExpr::new(children, phys_expr))
                as Arc<dyn PhysicalExpr>;
        // Simulate the partition value substitution that would happen in ParquetOpener
        let remapped_expr = dynamic_phys_expr
            .children()
            .into_iter()
            .map(|child_expr| {
                let Some(col_expr) = child_expr.downcast_ref::<phys_expr::Column>()
                else {
                    return Arc::clone(child_expr);
                };
                if col_expr.name() == "part" {
                    // simulate dynamic filter replacement with literal "A"
                    Arc::new(phys_expr::Literal::new(ScalarValue::Utf8(Some(
                        "A".to_string(),
                    )))) as Arc<dyn PhysicalExpr>
                } else {
                    Arc::clone(child_expr)
                }
            })
            .collect_vec();
        let dynamic_filter_expr =
            dynamic_phys_expr.with_new_children(remapped_expr).unwrap();
        // After substitution the expression is c1 > 5 AND part = "B" which should prune the file since the partition value is "A"
        let expected = &[false];
        let p =
            PruningPredicate::try_new(dynamic_filter_expr, Arc::clone(&schema)).unwrap();
        let result = p.prune(&statistics).unwrap();
        assert_eq!(result, expected);
    }

    #[test]
    fn row_group_predicate_lt_bool() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Boolean, false)]);
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_min@0 < true";

        // DF doesn't support arithmetic on boolean columns so
        // this predicate will error when evaluated
        let expr = col("c1").lt(lit(true));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_required_columns() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);
        let mut required_columns = RequiredColumns::new();
        // c1 < 1 and (c2 = 2 or c2 = 3)
        let expr = col("c1")
            .lt(lit(1))
            .and(col("c2").eq(lit(2)).or(col("c2").eq(lit(3))));
        let expected_expr = "c1_null_count@1 != row_count@2 AND c1_min@0 < 1 AND (c2_null_count@5 != row_count@2 AND c2_min@3 <= 2 AND 2 <= c2_max@4 OR c2_null_count@5 != row_count@2 AND c2_min@3 <= 3 AND 3 <= c2_max@4)";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut required_columns);
        assert_eq!(predicate_expr.to_string(), expected_expr);
        println!("required_columns: {required_columns:#?}"); // for debugging assertions below
        // c1 < 1 should add c1_min
        let c1_min_field = Field::new("c1_min", DataType::Int32, false);
        assert_eq!(
            required_columns.columns[0],
            (
                phys_expr::Column::new("c1", 0),
                StatisticsType::Min,
                c1_min_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        // c1 < 1 should add c1_null_count
        let c1_null_count_field = Field::new("c1_null_count", DataType::UInt64, false);
        assert_eq!(
            required_columns.columns[1],
            (
                phys_expr::Column::new("c1", 0),
                StatisticsType::NullCount,
                c1_null_count_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        // c1 < 1 should add row_count
        let row_count_field = Field::new("row_count", DataType::UInt64, false);
        assert_eq!(
            required_columns.columns[2],
            (
                phys_expr::Column::new("c1", 0),
                StatisticsType::RowCount,
                row_count_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        // c2 = 2 should add c2_min and c2_max
        let c2_min_field = Field::new("c2_min", DataType::Int32, false);
        assert_eq!(
            required_columns.columns[3],
            (
                phys_expr::Column::new("c2", 1),
                StatisticsType::Min,
                c2_min_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        let c2_max_field = Field::new("c2_max", DataType::Int32, false);
        assert_eq!(
            required_columns.columns[4],
            (
                phys_expr::Column::new("c2", 1),
                StatisticsType::Max,
                c2_max_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        // c2 = 2 should add c2_null_count
        let c2_null_count_field = Field::new("c2_null_count", DataType::UInt64, false);
        assert_eq!(
            required_columns.columns[5],
            (
                phys_expr::Column::new("c2", 1),
                StatisticsType::NullCount,
                c2_null_count_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        // c2 = 1 should add row_count
        let row_count_field = Field::new("row_count", DataType::UInt64, false);
        assert_eq!(
            required_columns.columns[2],
            (
                phys_expr::Column::new("c1", 0),
                StatisticsType::RowCount,
                row_count_field.with_nullable(true) // could be nullable if stats are not present
            )
        );
        // c2 = 3 shouldn't add any new statistics fields
        assert_eq!(required_columns.columns.len(), 6);

        Ok(())
    }

    #[test]
    fn row_group_predicate_in_list() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);
        // test c1 in(1, 2, 3)
        let expr = Expr::InList(InList::new(
            Box::new(col("c1")),
            vec![lit(1), lit(2), lit(3)],
            false,
        ));
        let expected_expr = "c1_null_count@2 != row_count@3 AND c1_min@0 <= 1 AND 1 <= c1_max@1 OR c1_null_count@2 != row_count@3 AND c1_min@0 <= 2 AND 2 <= c1_max@1 OR c1_null_count@2 != row_count@3 AND c1_min@0 <= 3 AND 3 <= c1_max@1";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_in_list_empty() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);
        // test c1 in()
        let expr = Expr::InList(InList::new(Box::new(col("c1")), vec![], false));
        let expected_expr = "true";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_in_list_negated() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);
        // test c1 not in(1, 2, 3)
        let expr = Expr::InList(InList::new(
            Box::new(col("c1")),
            vec![lit(1), lit(2), lit(3)],
            true,
        ));
        let expected_expr = "c1_null_count@2 != row_count@3 AND (c1_min@0 != 1 OR 1 != c1_max@1) AND c1_null_count@2 != row_count@3 AND (c1_min@0 != 2 OR 2 != c1_max@1) AND c1_null_count@2 != row_count@3 AND (c1_min@0 != 3 OR 3 != c1_max@1)";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_between() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);

        // test c1 BETWEEN 1 AND 5
        let expr1 = col("c1").between(lit(1), lit(5));

        // test 1 <= c1 <= 5
        let expr2 = col("c1").gt_eq(lit(1)).and(col("c1").lt_eq(lit(5)));

        let predicate_expr1 =
            test_build_predicate_expression(&expr1, &schema, &mut RequiredColumns::new());

        let predicate_expr2 =
            test_build_predicate_expression(&expr2, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr1.to_string(), predicate_expr2.to_string());

        Ok(())
    }

    #[test]
    fn row_group_predicate_between_with_in_list() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("c1", DataType::Int32, false),
            Field::new("c2", DataType::Int32, false),
        ]);
        // test c1 in(1, 2)
        let expr1 = col("c1").in_list(vec![lit(1), lit(2)], false);

        // test c2 BETWEEN 4 AND 5
        let expr2 = col("c2").between(lit(4), lit(5));

        // test c1 in(1, 2) and c2 BETWEEN 4 AND 5
        let expr3 = expr1.and(expr2);

        let expected_expr = "(c1_null_count@2 != row_count@3 AND c1_min@0 <= 1 AND 1 <= c1_max@1 OR c1_null_count@2 != row_count@3 AND c1_min@0 <= 2 AND 2 <= c1_max@1) AND c2_null_count@5 != row_count@3 AND c2_max@4 >= 4 AND c2_null_count@5 != row_count@3 AND c2_min@6 <= 5";
        let predicate_expr =
            test_build_predicate_expression(&expr3, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_in_list_to_many_values() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        // test c1 in(1..21)
        // in pruning.rs has MAX_LIST_VALUE_SIZE_REWRITE = 20, more than this value will be rewrite
        // always true
        let expr = col("c1").in_list((1..=21).map(lit).collect(), false);

        let expected_expr = "true";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_cast_int_int() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Int64) <= 1 AND 1 <= CAST(c1_max@1 AS Int64)";

        // test cast(c1 as int64) = 1
        // test column on the left
        let expr = cast(col("c1"), DataType::Int64).eq(lit(ScalarValue::Int64(Some(1))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(ScalarValue::Int64(Some(1))).eq(cast(col("c1"), DataType::Int64));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        let expected_expr =
            "c1_null_count@1 != row_count@2 AND TRY_CAST(c1_max@0 AS Int64) > 1";

        // test column on the left
        let expr =
            try_cast(col("c1"), DataType::Int64).gt(lit(ScalarValue::Int64(Some(1))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr =
            lit(ScalarValue::Int64(Some(1))).lt(try_cast(col("c1"), DataType::Int64));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_cast_string_string() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Utf8View, false)]);
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Utf8) <= 1 AND 1 <= CAST(c1_max@1 AS Utf8)";

        // test column on the left
        let expr = cast(col("c1"), DataType::Utf8)
            .eq(lit(ScalarValue::Utf8(Some("1".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(ScalarValue::Utf8(Some("1".to_string())))
            .eq(cast(col("c1"), DataType::Utf8));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_cast_string_int() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Utf8View, false)]);
        let expected_expr = "true";

        // test column on the left
        let expr = cast(col("c1"), DataType::Int32).eq(lit(ScalarValue::Int32(Some(1))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(ScalarValue::Int32(Some(1))).eq(cast(col("c1"), DataType::Int32));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_cast_int_string() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        let expected_expr = "true";

        // test column on the left
        let expr = cast(col("c1"), DataType::Utf8)
            .eq(lit(ScalarValue::Utf8(Some("1".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(ScalarValue::Utf8(Some("1".to_string())))
            .eq(cast(col("c1"), DataType::Utf8));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_date_date() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Date32, false)]);
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Date64) <= 1970-01-01 AND 1970-01-01 <= CAST(c1_max@1 AS Date64)";

        // test column on the left
        let expr =
            cast(col("c1"), DataType::Date64).eq(lit(ScalarValue::Date64(Some(123))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr =
            lit(ScalarValue::Date64(Some(123))).eq(cast(col("c1"), DataType::Date64));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_dict_string_date() -> Result<()> {
        // Test with Dictionary<UInt8, Utf8> for the literal
        let schema = Schema::new(vec![Field::new("c1", DataType::Date32, false)]);
        let expected_expr = "true";

        // test column on the left
        let expr = cast(
            col("c1"),
            DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Utf8)),
        )
        .eq(lit(ScalarValue::Utf8(Some("2024-01-01".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(ScalarValue::Utf8(Some("2024-01-01".to_string()))).eq(cast(
            col("c1"),
            DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Utf8)),
        ));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_date_dict_string() -> Result<()> {
        // Test with Dictionary<UInt8, Utf8> for the column
        let schema = Schema::new(vec![Field::new(
            "c1",
            DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Utf8)),
            false,
        )]);
        let expected_expr = "true";

        // test column on the left
        let expr =
            cast(col("c1"), DataType::Date32).eq(lit(ScalarValue::Date32(Some(123))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr =
            lit(ScalarValue::Date32(Some(123))).eq(cast(col("c1"), DataType::Date32));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_dict_dict_same_value_type() -> Result<()> {
        // Test with Dictionary types that have the same value type but different key types
        let schema = Schema::new(vec![Field::new(
            "c1",
            DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Utf8)),
            false,
        )]);

        // Direct comparison with no cast
        let expr = col("c1").eq(lit(ScalarValue::Utf8(Some("test".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        let expected_expr =
            "c1_null_count@2 != row_count@3 AND c1_min@0 <= test AND test <= c1_max@1";
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // Test with column cast to a dictionary with different key type
        let expr = cast(
            col("c1"),
            DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Utf8)),
        )
        .eq(lit(ScalarValue::Utf8(Some("test".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Dictionary(UInt16, Utf8)) <= test AND test <= CAST(c1_max@1 AS Dictionary(UInt16, Utf8))";
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_dict_dict_different_value_type() -> Result<()> {
        // Test with Dictionary types that have different value types
        let schema = Schema::new(vec![Field::new(
            "c1",
            DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Int32)),
            false,
        )]);
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Int64) <= 123 AND 123 <= CAST(c1_max@1 AS Int64)";

        // Test with literal of a different type
        let expr =
            cast(col("c1"), DataType::Int64).eq(lit(ScalarValue::Int64(Some(123))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_nested_dict() -> Result<()> {
        // Test with nested Dictionary types
        let schema = Schema::new(vec![Field::new(
            "c1",
            DataType::Dictionary(
                Box::new(DataType::UInt8),
                Box::new(DataType::Dictionary(
                    Box::new(DataType::UInt16),
                    Box::new(DataType::Utf8),
                )),
            ),
            false,
        )]);
        let expected_expr =
            "c1_null_count@2 != row_count@3 AND c1_min@0 <= test AND test <= c1_max@1";

        // Test with a simple literal
        let expr = col("c1").eq(lit(ScalarValue::Utf8(Some("test".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_dict_date_dict_date() -> Result<()> {
        // Test with dictionary-wrapped date types for both sides
        let schema = Schema::new(vec![Field::new(
            "c1",
            DataType::Dictionary(Box::new(DataType::UInt8), Box::new(DataType::Date32)),
            false,
        )]);
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Dictionary(UInt16, Date64)) <= 1970-01-01 AND 1970-01-01 <= CAST(c1_max@1 AS Dictionary(UInt16, Date64))";

        // Test with a cast to a different date type
        let expr = cast(
            col("c1"),
            DataType::Dictionary(Box::new(DataType::UInt16), Box::new(DataType::Date64)),
        )
        .eq(lit(ScalarValue::Date64(Some(123))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_date_string() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Utf8, false)]);
        let expected_expr = "true";

        // test column on the left
        let expr =
            cast(col("c1"), DataType::Date32).eq(lit(ScalarValue::Date32(Some(123))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr =
            lit(ScalarValue::Date32(Some(123))).eq(cast(col("c1"), DataType::Date32));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_string_date() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Date32, false)]);
        let expected_expr = "true";

        // test column on the left
        let expr = cast(col("c1"), DataType::Utf8)
            .eq(lit(ScalarValue::Utf8(Some("2024-01-01".to_string()))));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        // test column on the right
        let expr = lit(ScalarValue::Utf8(Some("2024-01-01".to_string())))
            .eq(cast(col("c1"), DataType::Utf8));
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn row_group_predicate_cast_list() -> Result<()> {
        let schema = Schema::new(vec![Field::new("c1", DataType::Int32, false)]);
        // test cast(c1 as int64) in int64(1, 2, 3)
        let expr = Expr::InList(InList::new(
            Box::new(cast(col("c1"), DataType::Int64)),
            vec![
                lit(ScalarValue::Int64(Some(1))),
                lit(ScalarValue::Int64(Some(2))),
                lit(ScalarValue::Int64(Some(3))),
            ],
            false,
        ));
        let expected_expr = "c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Int64) <= 1 AND 1 <= CAST(c1_max@1 AS Int64) OR c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Int64) <= 2 AND 2 <= CAST(c1_max@1 AS Int64) OR c1_null_count@2 != row_count@3 AND CAST(c1_min@0 AS Int64) <= 3 AND 3 <= CAST(c1_max@1 AS Int64)";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        let expr = Expr::InList(InList::new(
            Box::new(cast(col("c1"), DataType::Int64)),
            vec![
                lit(ScalarValue::Int64(Some(1))),
                lit(ScalarValue::Int64(Some(2))),
                lit(ScalarValue::Int64(Some(3))),
            ],
            true,
        ));
        let expected_expr = "c1_null_count@2 != row_count@3 AND (CAST(c1_min@0 AS Int64) != 1 OR 1 != CAST(c1_max@1 AS Int64)) AND c1_null_count@2 != row_count@3 AND (CAST(c1_min@0 AS Int64) != 2 OR 2 != CAST(c1_max@1 AS Int64)) AND c1_null_count@2 != row_count@3 AND (CAST(c1_min@0 AS Int64) != 3 OR 3 != CAST(c1_max@1 AS Int64))";
        let predicate_expr =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        assert_eq!(predicate_expr.to_string(), expected_expr);

        Ok(())
    }

    #[test]
    fn prune_decimal_data() {
        // decimal(9,2)
        let schema = Arc::new(Schema::new(vec![Field::new(
            "s1",
            DataType::Decimal128(9, 2),
            true,
        )]));

        prune_with_expr(
            // s1 > 5
            col("s1").gt(lit(ScalarValue::Decimal128(Some(500), 9, 2))),
            &schema,
            // If the data is written by spark, the physical data type is INT32 in the parquet
            // So we use the INT32 type of statistic.
            &TestStatistics::new().with(
                "s1",
                ContainerStats::new_i32(
                    vec![Some(0), Some(4), None, Some(3)], // min
                    vec![Some(5), Some(6), Some(4), None], // max
                ),
            ),
            &[false, true, false, true],
        );

        prune_with_expr(
            // with cast column to other type
            cast(col("s1"), DataType::Decimal128(14, 3))
                .gt(lit(ScalarValue::Decimal128(Some(5000), 14, 3))),
            &schema,
            &TestStatistics::new().with(
                "s1",
                ContainerStats::new_i32(
                    vec![Some(0), Some(4), None, Some(3)], // min
                    vec![Some(5), Some(6), Some(4), None], // max
                ),
            ),
            &[false, true, false, true],
        );

        prune_with_expr(
            // with try cast column to other type
            try_cast(col("s1"), DataType::Decimal128(14, 3))
                .gt(lit(ScalarValue::Decimal128(Some(5000), 14, 3))),
            &schema,
            &TestStatistics::new().with(
                "s1",
                ContainerStats::new_i32(
                    vec![Some(0), Some(4), None, Some(3)], // min
                    vec![Some(5), Some(6), Some(4), None], // max
                ),
            ),
            &[false, true, false, true],
        );

        // decimal(18,2)
        let schema = Arc::new(Schema::new(vec![Field::new(
            "s1",
            DataType::Decimal128(18, 2),
            true,
        )]));
        prune_with_expr(
            // s1 > 5
            col("s1").gt(lit(ScalarValue::Decimal128(Some(500), 18, 2))),
            &schema,
            // If the data is written by spark, the physical data type is INT64 in the parquet
            // So we use the INT32 type of statistic.
            &TestStatistics::new().with(
                "s1",
                ContainerStats::new_i64(
                    vec![Some(0), Some(4), None, Some(3)], // min
                    vec![Some(5), Some(6), Some(4), None], // max
                ),
            ),
            &[false, true, false, true],
        );

        // decimal(23,2)
        let schema = Arc::new(Schema::new(vec![Field::new(
            "s1",
            DataType::Decimal128(23, 2),
            true,
        )]));

        prune_with_expr(
            // s1 > 5
            col("s1").gt(lit(ScalarValue::Decimal128(Some(500), 23, 2))),
            &schema,
            &TestStatistics::new().with(
                "s1",
                ContainerStats::new_decimal128(
                    vec![Some(0), Some(400), None, Some(300)], // min
                    vec![Some(500), Some(600), Some(400), None], // max
                    23,
                    2,
                ),
            ),
            &[false, true, false, true],
        );
    }

    #[test]
    fn prune_api() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("s1", DataType::Utf8, true),
            Field::new("s2", DataType::Int32, true),
        ]));

        let statistics = TestStatistics::new().with(
            "s2",
            ContainerStats::new_i32(
                vec![Some(0), Some(4), None, Some(3)], // min
                vec![Some(5), Some(6), None, None],    // max
            ),
        );
        prune_with_expr(
            // Prune using s2 > 5
            col("s2").gt(lit(5)),
            &schema,
            &statistics,
            // s2 [0, 5] ==> no rows should pass
            // s2 [4, 6] ==> some rows could pass
            // No stats for s2 ==> some rows could pass
            // s2 [3, None] (null max) ==> some rows could pass
            &[false, true, true, true],
        );

        prune_with_expr(
            // filter with cast
            cast(col("s2"), DataType::Int64).gt(lit(ScalarValue::Int64(Some(5)))),
            &schema,
            &statistics,
            &[false, true, true, true],
        );
    }

    #[test]
    fn prune_not_eq_data() {
        let schema = Arc::new(Schema::new(vec![Field::new("s1", DataType::Utf8, true)]));

        prune_with_expr(
            // Prune using s2 != 'M'
            col("s1").not_eq(lit("M")),
            &schema,
            &TestStatistics::new().with(
                "s1",
                ContainerStats::new_utf8(
                    vec![Some("A"), Some("A"), Some("N"), Some("M"), None, Some("A")], // min
                    vec![Some("Z"), Some("L"), Some("Z"), Some("M"), None, None], // max
                ),
            ),
            // s1 [A, Z] ==> might have values that pass predicate
            // s1 [A, L] ==> all rows pass the predicate
            // s1 [N, Z] ==> all rows pass the predicate
            // s1 [M, M] ==> all rows do not pass the predicate
            // No stats for s2 ==> some rows could pass
            // s2 [3, None] (null max) ==> some rows could pass
            &[true, true, true, false, true, true],
        );
    }

    /// Creates setup for boolean chunk pruning
    ///
    /// For predicate "b1" (boolean expr)
    /// b1 [false, false] ==> no rows can pass (not keep)
    /// b1 [false, true] ==> some rows could pass (must keep)
    /// b1 [true, true] ==> all rows must pass (must keep)
    /// b1 [NULL, NULL]  ==> unknown (must keep)
    /// b1 [false, NULL]  ==> unknown (must keep)
    ///
    /// For predicate "!b1" (boolean expr)
    /// b1 [false, false] ==> all rows pass (must keep)
    /// b1 [false, true] ==> some rows could pass (must keep)
    /// b1 [true, true] ==> no rows can pass (not keep)
    /// b1 [NULL, NULL]  ==> unknown (must keep)
    /// b1 [false, NULL]  ==> unknown (must keep)
    fn bool_setup() -> (SchemaRef, TestStatistics, Vec<bool>, Vec<bool>) {
        let schema =
            Arc::new(Schema::new(vec![Field::new("b1", DataType::Boolean, true)]));

        let statistics = TestStatistics::new().with(
            "b1",
            ContainerStats::new_bool(
                vec![Some(false), Some(false), Some(true), None, Some(false)], // min
                vec![Some(false), Some(true), Some(true), None, None],         // max
            ),
        );
        let expected_true = vec![false, true, true, true, true];
        let expected_false = vec![true, true, false, true, true];

        (schema, statistics, expected_true, expected_false)
    }

    #[test]
    fn prune_bool_const_expr() {
        let (schema, statistics, _, _) = bool_setup();

        prune_with_expr(
            // true
            lit(true),
            &schema,
            &statistics,
            &[true, true, true, true, true],
        );

        prune_with_expr(
            // false
            lit(false),
            &schema,
            &statistics,
            &[false, false, false, false, false],
        );
    }

    #[test]
    fn prune_bool_column() {
        let (schema, statistics, expected_true, _) = bool_setup();

        prune_with_expr(
            // b1
            col("b1"),
            &schema,
            &statistics,
            &expected_true,
        );
    }

    #[test]
    fn prune_bool_not_column() {
        let (schema, statistics, _, expected_false) = bool_setup();

        prune_with_expr(
            // !b1
            col("b1").not(),
            &schema,
            &statistics,
            &expected_false,
        );
    }

    #[test]
    fn prune_bool_column_eq_true() {
        let (schema, statistics, expected_true, _) = bool_setup();

        prune_with_expr(
            // b1 = true
            col("b1").eq(lit(true)),
            &schema,
            &statistics,
            &expected_true,
        );
    }

    #[test]
    fn prune_bool_not_column_eq_true() {
        let (schema, statistics, _, expected_false) = bool_setup();

        prune_with_expr(
            // !b1 = true
            col("b1").not().eq(lit(true)),
            &schema,
            &statistics,
            &expected_false,
        );
    }

    /// Creates a setup for chunk pruning, modeling a int32 column "i"
    /// with 5 different containers (e.g. RowGroups). They have [min,
    /// max]:
    ///
    /// i [-5, 5]
    /// i [1, 11]
    /// i [-11, -1]
    /// i [NULL, NULL]
    /// i [1, NULL]
    fn int32_setup() -> (SchemaRef, TestStatistics) {
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));

        let statistics = TestStatistics::new().with(
            "i",
            ContainerStats::new_i32(
                vec![Some(-5), Some(1), Some(-11), None, Some(1)], // min
                vec![Some(5), Some(11), Some(-1), None, None],     // max
            ),
        );
        (schema, statistics)
    }

    #[test]
    fn prune_int32_col_gt_zero() {
        let (schema, statistics) = int32_setup();

        // Expression "i > 0" and "-i < 0"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> all rows must pass (must keep)
        // i [-11, -1] ==>  no rows can pass (not keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> unknown (must keep)
        let expected_ret = &[true, true, false, true, true];

        // i > 0
        prune_with_expr(col("i").gt(lit(0)), &schema, &statistics, expected_ret);

        // -i < 0
        prune_with_expr(
            Expr::Negative(Box::new(col("i"))).lt(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_lte_zero() {
        let (schema, statistics) = int32_setup();

        // Expression "i <= 0" and "-i >= 0"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> no rows can pass (not keep)
        // i [-11, -1] ==>  all rows must pass (must keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> no rows can pass (not keep)
        let expected_ret = &[true, false, true, true, false];

        prune_with_expr(
            // i <= 0
            col("i").lt_eq(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // -i >= 0
            Expr::Negative(Box::new(col("i"))).gt_eq(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_lte_zero_cast() {
        let (schema, statistics) = int32_setup();

        // Expression "cast(i as utf8) <= '0'"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> no rows can pass in theory, -0.22 (conservatively keep)
        // i [-11, -1] ==>  no rows could pass in theory (conservatively keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> no rows can pass (conservatively keep)
        let expected_ret = &[true, true, true, true, true];

        prune_with_expr(
            // cast(i as utf8) <= 0
            cast(col("i"), DataType::Utf8).lt_eq(lit("0")),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // try_cast(i as utf8) <= 0
            try_cast(col("i"), DataType::Utf8).lt_eq(lit("0")),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // cast(-i as utf8) >= 0
            cast(Expr::Negative(Box::new(col("i"))), DataType::Utf8).gt_eq(lit("0")),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // try_cast(-i as utf8) >= 0
            try_cast(Expr::Negative(Box::new(col("i"))), DataType::Utf8).gt_eq(lit("0")),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_eq_zero() {
        let (schema, statistics) = int32_setup();

        // Expression "i = 0"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> no rows can pass (not keep)
        // i [-11, -1] ==>  no rows can pass (not keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> no rows can pass (not keep)
        let expected_ret = &[true, false, false, true, false];

        prune_with_expr(
            // i = 0
            col("i").eq(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_is_not_distinct_from() {
        let (schema, statistics) = int32_setup();

        // Without null counts, IS NOT DISTINCT FROM a non-null literal can
        // still use min/max ranges, but unknown all-null containers must be kept.
        let expected_ret = &[true, false, false, true, false];

        prune_with_expr(
            is_not_distinct_from(col("i"), lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );

        // The operator is symmetric, so the scalar-left form should prune the
        // same row groups.
        prune_with_expr(
            is_not_distinct_from(lit(0), col("i")),
            &schema,
            &statistics,
            expected_ret,
        );

        let statistics = statistics
            .with_row_counts("i", vec![Some(10), Some(9), None, Some(4), Some(10)])
            .with_null_counts("i", vec![Some(0), Some(1), None, Some(4), Some(0)]);

        let expected_ret = &[true, false, false, false, false];
        prune_with_expr(
            is_not_distinct_from(col("i"), lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );

        let expected_ret = &[false, true, true, true, false];
        prune_with_expr(
            is_not_distinct_from(col("i"), lit(ScalarValue::Int32(None))),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_is_distinct_from() {
        let schema = Arc::new(Schema::new(vec![Field::new("i", DataType::Int32, true)]));
        let statistics = TestStatistics::new().with(
            "i",
            ContainerStats::new_i32(
                vec![Some(0), Some(0), Some(5), None],
                vec![Some(0), Some(2), Some(5), None],
            )
            .with_row_counts(vec![Some(2), Some(2), Some(2), Some(2)])
            .with_null_counts(vec![Some(0), Some(0), Some(0), Some(2)]),
        );

        let expected_ret = &[false, true, true, true];
        prune_with_expr(
            is_distinct_from(col("i"), lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );

        // The operator is symmetric, so the scalar-left form should prune the
        // same row groups.
        prune_with_expr(
            is_distinct_from(lit(0), col("i")),
            &schema,
            &statistics,
            expected_ret,
        );

        let expected_ret = &[true, true, true, false];
        prune_with_expr(
            is_distinct_from(col("i"), lit(ScalarValue::Int32(None))),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_eq_zero_cast() {
        let (schema, statistics) = int32_setup();

        // Expression "cast(i as int64) = 0"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> no rows can pass (not keep)
        // i [-11, -1] ==>  no rows can pass (not keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> no rows can pass (not keep)
        let expected_ret = &[true, false, false, true, false];

        prune_with_expr(
            cast(col("i"), DataType::Int64).eq(lit(0i64)),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            try_cast(col("i"), DataType::Int64).eq(lit(0i64)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_eq_zero_cast_as_str() {
        let (schema, statistics) = int32_setup();

        // Note the cast is to a string where sorting properties are
        // not the same as integers
        //
        // Expression "cast(i as utf8) = '0'"
        // i [-5, 5] ==> some rows could pass (keep)
        // i [1, 11] ==> no rows can pass  (could keep)
        // i [-11, -1] ==>  no rows can pass (could keep)
        // i [NULL, NULL]  ==> unknown (keep)
        // i [1, NULL]  ==> no rows can pass (could keep)
        let expected_ret = &[true, true, true, true, true];

        prune_with_expr(
            cast(col("i"), DataType::Utf8).eq(lit("0")),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_col_lt_neg_one() {
        let (schema, statistics) = int32_setup();

        // Expression "i > -1" and "-i < 1"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> all rows must pass (must keep)
        // i [-11, -1] ==>  no rows can pass (not keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> all rows must pass (must keep)
        let expected_ret = &[true, true, false, true, true];

        prune_with_expr(
            // i > -1
            col("i").gt(lit(-1)),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // -i < 1
            Expr::Negative(Box::new(col("i"))).lt(lit(1)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_is_null() {
        let (schema, statistics) = int32_setup();

        // Expression "i IS NULL" when there are no null statistics,
        // should all be kept
        let expected_ret = &[true, true, true, true, true];

        prune_with_expr(
            // i IS NULL, no null statistics
            col("i").is_null(),
            &schema,
            &statistics,
            expected_ret,
        );

        // provide null counts for each column
        let statistics = statistics.with_null_counts(
            "i",
            vec![
                Some(0), // no nulls (don't keep)
                Some(1), // 1 null
                None,    // unknown nulls
                None, // unknown nulls (min/max are both null too, like no stats at all)
                Some(0), // 0 nulls (max=null too which means no known max) (don't keep)
            ],
        );

        let expected_ret = &[false, true, true, true, false];

        prune_with_expr(
            // i IS NULL, with actual null statistics
            col("i").is_null(),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_int32_column_is_known_all_null() {
        let (schema, statistics) = int32_setup();

        // Expression "i < 0"
        // i [-5, 5] ==> some rows could pass (must keep)
        // i [1, 11] ==> no rows can pass (not keep)
        // i [-11, -1] ==>  all rows must pass (must keep)
        // i [NULL, NULL]  ==> unknown (must keep)
        // i [1, NULL]  ==> no rows can pass (not keep)
        let expected_ret = &[true, false, true, true, false];

        prune_with_expr(
            // i < 0
            col("i").lt(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );

        // provide row counts for each column
        let statistics = statistics.with_row_counts(
            "i",
            vec![
                Some(10), // 10 rows of data
                Some(9),  // 9 rows of data
                None,     // unknown row counts
                Some(4),
                Some(10),
            ],
        );

        // pruning result is still the same if we only know row counts
        prune_with_expr(
            // i < 0, with only row counts statistics
            col("i").lt(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );

        // provide null counts for each column
        let statistics = statistics.with_null_counts(
            "i",
            vec![
                Some(0), // no nulls
                Some(1), // 1 null
                None,    // unknown nulls
                Some(4), // 4 nulls, which is the same as the row counts, i.e. this column is all null (don't keep)
                Some(0), // 0 nulls (max=null too which means no known max)
            ],
        );

        // Expression "i < 0" with actual null and row counts statistics
        // col | min, max     | row counts | null counts |
        // ----+--------------+------------+-------------+
        //  i  | [-5, 5]      | 10         | 0           | ==> Some rows could pass (must keep)
        //  i  | [1, 11]      | 9          | 1           | ==> No rows can pass (not keep)
        //  i  | [-11,-1]     | Unknown    | Unknown     | ==> All rows must pass (must keep)
        //  i  | [NULL, NULL] | 4          | 4           | ==> The column is all null (not keep)
        //  i  | [1, NULL]    | 10         | 0           | ==> No rows can pass (not keep)
        let expected_ret = &[true, false, true, false, false];

        prune_with_expr(
            // i < 0, with actual null and row counts statistics
            col("i").lt(lit(0)),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn prune_cast_scalar() {
        // The data type of column i is INT32
        let (schema, statistics) = int32_setup();
        let expected_ret = &[true, true, false, true, true];

        prune_with_expr(
            // i > int64(0)
            col("i").gt(cast(lit(ScalarValue::Int64(Some(0))), DataType::Int32)),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // cast(i as int64) > int64(0)
            cast(col("i"), DataType::Int64).gt(lit(ScalarValue::Int64(Some(0)))),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // try_cast(i as int64) > int64(0)
            try_cast(col("i"), DataType::Int64).gt(lit(ScalarValue::Int64(Some(0)))),
            &schema,
            &statistics,
            expected_ret,
        );

        prune_with_expr(
            // `-cast(i as int64) < 0` convert to `cast(i as int64) > -0`
            Expr::Negative(Box::new(cast(col("i"), DataType::Int64)))
                .lt(lit(ScalarValue::Int64(Some(0)))),
            &schema,
            &statistics,
            expected_ret,
        );
    }

    #[test]
    fn test_increment_utf8() {
        // Basic ASCII
        assert_eq!(increment_utf8("abc").unwrap(), "abd");
        assert_eq!(increment_utf8("abz").unwrap(), "ab{");

        // Test around ASCII 127 (DEL)
        assert_eq!(increment_utf8("~").unwrap(), "\u{7f}"); // 126 -> 127
        assert_eq!(increment_utf8("\u{7f}").unwrap(), "\u{80}"); // 127 -> 128

        // Test 2-byte UTF-8 sequences
        assert_eq!(increment_utf8("ß").unwrap(), "à"); // U+00DF -> U+00E0

        // Test 3-byte UTF-8 sequences
        assert_eq!(increment_utf8("℣").unwrap(), "ℤ"); // U+2123 -> U+2124

        // Test at UTF-8 boundaries
        assert_eq!(increment_utf8("\u{7FF}").unwrap(), "\u{800}"); // 2-byte to 3-byte boundary
        assert_eq!(increment_utf8("\u{FFFF}").unwrap(), "\u{10000}"); // 3-byte to 4-byte boundary

        // Test that if we can't increment we return None
        assert!(increment_utf8("").is_none());
        assert!(increment_utf8("\u{10FFFF}").is_none()); // U+10FFFF is the max code point

        // Test that if we can't increment the last character we do the previous one and truncate
        assert_eq!(increment_utf8("a\u{10FFFF}").unwrap(), "b");

        // Test surrogate pair range (0xD800..=0xDFFF)
        assert_eq!(increment_utf8("a\u{D7FF}").unwrap(), "b");
        assert!(increment_utf8("\u{D7FF}").is_none());

        // Test non-characters range (0xFDD0..=0xFDEF)
        assert_eq!(increment_utf8("a\u{FDCF}").unwrap(), "b");
        assert!(increment_utf8("\u{FDCF}").is_none());

        // Test private use area limit (>= 0x110000)
        assert_eq!(increment_utf8("a\u{10FFFF}").unwrap(), "b");
        assert!(increment_utf8("\u{10FFFF}").is_none()); // Can't increment past max valid codepoint
    }

    /// Creates a setup for chunk pruning, modeling a utf8 column "s1"
    /// with 5 different containers (e.g. RowGroups). They have [min,
    /// max]:
    /// s1 ["A", "Z"]
    /// s1 ["A", "L"]
    /// s1 ["N", "Z"]
    /// s1 [NULL, NULL]
    /// s1 ["A", NULL]
    /// s1 ["", "A"]
    /// s1 ["", ""]
    /// s1 ["AB", "A\u{10ffff}"]
    /// s1 ["A\u{10ffff}\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]
    fn utf8_setup() -> (SchemaRef, TestStatistics) {
        let schema = Arc::new(Schema::new(vec![Field::new("s1", DataType::Utf8, true)]));

        let statistics = TestStatistics::new().with(
            "s1",
            ContainerStats::new_utf8(
                vec![
                    Some("A"),
                    Some("A"),
                    Some("N"),
                    Some("M"),
                    None,
                    Some("A"),
                    Some(""),
                    Some(""),
                    Some("AB"),
                    Some("A\u{10ffff}\u{10ffff}"),
                ], // min
                vec![
                    Some("Z"),
                    Some("L"),
                    Some("Z"),
                    Some("M"),
                    None,
                    None,
                    Some("A"),
                    Some(""),
                    Some("A\u{10ffff}\u{10ffff}\u{10ffff}"),
                    Some("A\u{10ffff}\u{10ffff}"),
                ], // max
            ),
        );
        (schema, statistics)
    }

    #[test]
    fn prune_utf8_eq() {
        let (schema, statistics) = utf8_setup();

        let expr = col("s1").eq(lit("A"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["M", "M"] ==> no rows can pass (not keep)
            false,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> no rows can pass (not keep)
            false,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").eq(lit(""));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["A", "L"] ==> no rows can pass (not keep)
            false,
            // s1 ["N", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["M", "M"] ==> no rows can pass (not keep)
            false,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> no rows can pass (not keep)
            false,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> all rows must pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_utf8_not_eq() {
        let (schema, statistics) = utf8_setup();

        let expr = col("s1").not_eq(lit("A"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["M", "M"] ==> all rows must pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> all rows must pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").not_eq(lit(""));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["A", "L"] ==> all rows must pass (must keep)
            true,
            // s1 ["N", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["M", "M"] ==> all rows must pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> no rows can pass (not keep)
            false,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_utf8_like_one() {
        let (schema, statistics) = utf8_setup();

        let expr = col("s1").like(lit("A_"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["M", "M"] ==> no rows can pass (not keep)
            false,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> no rows can pass (not keep)
            false,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit("_A_"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["M", "M"] ==> some rows could pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> some rows could pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit("_"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["A", "L"] ==> all rows must pass (must keep)
            true,
            // s1 ["N", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["M", "M"] ==> all rows must pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> all rows must pass (must keep)
            true,
            // s1 ["", ""]  ==> all rows must pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit(""));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["A", "L"] ==> no rows can pass (not keep)
            false,
            // s1 ["N", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["M", "M"] ==> no rows can pass (not keep)
            false,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> no rows can pass (not keep)
            false,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> all rows must pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_utf8_like_many() {
        let (schema, statistics) = utf8_setup();

        let expr = col("s1").like(lit("A%"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["M", "M"] ==> no rows can pass (not keep)
            false,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> no rows can pass (not keep)
            false,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit("%A%"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["M", "M"] ==> some rows could pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> some rows could pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit("%"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["A", "L"] ==> all rows must pass (must keep)
            true,
            // s1 ["N", "Z"] ==> all rows must pass (must keep)
            true,
            // s1 ["M", "M"] ==> all rows must pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> unknown (must keep)
            true,
            // s1 ["", "A"]  ==> all rows must pass (must keep)
            true,
            // s1 ["", ""]  ==> all rows must pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> all rows must pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit(""));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["A", "L"] ==> no rows can pass (not keep)
            false,
            // s1 ["N", "Z"] ==> no rows can pass (not keep)
            false,
            // s1 ["M", "M"] ==> no rows can pass (not keep)
            false,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> no rows can pass (not keep)
            false,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> all rows must pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    // `build_like_match()` must honor `\` escapes when scanning the pattern for
    // wildcards.
    #[test]
    fn prune_utf8_like_escaped_chars() {
        let schema = Arc::new(Schema::new(vec![Field::new("s1", DataType::Utf8, true)]));
        let statistics = TestStatistics::new().with(
            "s1",
            ContainerStats::new_utf8(
                vec![
                    Some("foo_aaa"),
                    Some(r#"foo\aaa"#),
                    Some("foo"),
                    Some("bar"),
                    Some("foo%aaa"),
                    Some("%foo_aaa"),
                ], // min
                vec![
                    Some("foo_zzz"),
                    Some(r#"foo\zzz"#),
                    Some("foozzz"),
                    Some("baz"),
                    Some("foo%zzz"),
                    Some("%foo_zzz"),
                ], // max
            ),
        );

        let expr = col("s1").like(lit(r#"foo\_%"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => every value starts with literal
            // "foo_" and matches the pattern; must keep.
            true,
            // s1 ["foo\aaa", "foo\zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo", "foozzz"] => stats don't prove "foo_" is or isn't in
            // range; must conservatively keep.
            true,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => no rows can pass (not keep)
            false,
            // s1 ["%foo_aaa", "%foo_zzz"] => no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit(r#"foo\\%"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo\aaa", "foo\zzz"] => every value starts with literal
            // "foo\" and matches the pattern; must keep.
            true,
            // s1 ["foo", "foozzz"] => stats don't prove "foo\" is or isn't in
            // range; must conservatively keep.
            true,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => no rows can pass (not keep)
            false,
            // s1 ["%foo_aaa", "%foo_zzz"] => no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").like(lit(r#"foo\%%"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo\aaa", "foo\zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo", "foozzz"] => range straddles "foo%"; must keep.
            true,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => every value starts with literal
            // "foo%" and matches the pattern; must keep.
            true,
            // s1 ["%foo_aaa", "%foo_zzz"] => no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        // No wildcard after escapes: pattern reduces to an equality check on
        // the literal "foo_".
        let expr = col("s1").like(lit(r#"foo\_"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo\aaa", "foo\zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo", "foozzz"] => "foo_" is within the range; must keep.
            true,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => no rows can pass (not keep)
            false,
            // s1 ["%foo_aaa", "%foo_zzz"] => no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        // Leading escaped `%`: prefix is "%foo" (non-empty), so the guard
        // for "all wildcards" must NOT bail out here.
        let expr = col("s1").like(lit(r#"\%foo%"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo\aaa", "foo\zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo", "foozzz"] => no rows can pass (not keep)
            false,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => no rows can pass (not keep)
            false,
            // s1 ["%foo_aaa", "%foo_zzz"] => every value starts with literal
            // "%foo" and matches the pattern; must keep.
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        // Two escaped wildcards, no real wildcard: equality on "foo%_".
        let expr = col("s1").like(lit(r#"foo\%\_"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo\aaa", "foo\zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo", "foozzz"] => "foo%_" is within the range; must keep.
            true,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => no rows can pass (not keep)
            false,
            // s1 ["%foo_aaa", "%foo_zzz"] => no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        // Escaped backslash followed by more literal chars before the
        // wildcard: prefix is "foo\bar".
        let expr = col("s1").like(lit(r#"foo\\bar%"#));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["foo_aaa", "foo_zzz"] => no rows can pass (not keep)
            false,
            // s1 ["foo\aaa", "foo\zzz"] => range straddles "foo\bar"; must
            // keep.
            true,
            // s1 ["foo", "foozzz"] => range straddles "foo\bar"; must keep.
            true,
            // s1 ["bar", "baz"] => no rows can pass (not keep)
            false,
            // s1 ["foo%aaa", "foo%zzz"] => no rows can pass (not keep)
            false,
            // s1 ["%foo_aaa", "%foo_zzz"] => no rows can pass (not keep)
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_utf8_not_like_one() {
        let (schema, statistics) = utf8_setup();

        let expr = col("s1").not_like(lit("A\u{10ffff}_"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["M", "M"] ==> some rows could pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> some rows could pass (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> some rows could pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> no row match. (min, max) maybe truncate
            // original (min, max) maybe ("A\u{10ffff}\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}\u{10ffff}\u{10ffff}")
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    #[test]
    fn prune_utf8_not_like_many() {
        let (schema, statistics) = utf8_setup();

        let expr = col("s1").not_like(lit("A\u{10ffff}%"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["M", "M"] ==> some rows could pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> some rows could pass (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> some rows could pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> no row match
            false,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").not_like(lit("A\u{10ffff}%\u{10ffff}"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["M", "M"] ==> some rows could pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> some rows could pass (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> some rows could pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").not_like(lit("A\u{10ffff}%\u{10ffff}_"));
        #[rustfmt::skip]
        let expected_ret = &[
            // s1 ["A", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["A", "L"] ==> some rows could pass (must keep)
            true,
            // s1 ["N", "Z"] ==> some rows could pass (must keep)
            true,
            // s1 ["M", "M"] ==> some rows could pass (must keep)
            true,
            // s1 [NULL, NULL]  ==> unknown (must keep)
            true,
            // s1 ["A", NULL]  ==> some rows could pass (must keep)
            true,
            // s1 ["", "A"]  ==> some rows could pass (must keep)
            true,
            // s1 ["", ""]  ==> some rows could pass (must keep)
            true,
            // s1 ["AB", "A\u{10ffff}\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
            // s1 ["A\u{10ffff}\u{10ffff}", "A\u{10ffff}\u{10ffff}"]  ==> some rows could pass (must keep)
            true,
        ];
        prune_with_expr(expr, &schema, &statistics, expected_ret);

        let expr = col("s1").not_like(lit("A\\%%"));
        let statistics = TestStatistics::new().with(
            "s1",
            ContainerStats::new_utf8(
                vec![Some("A%a"), Some("A")],
                vec![Some("A%c"), Some("A")],
            ),
        );
        let expected_ret = &[false, true];
        prune_with_expr(expr, &schema, &statistics, expected_ret);
    }

    #[test]
    fn test_rewrite_expr_to_prunable() {
        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
        let df_schema = DFSchema::try_from(schema.clone()).unwrap();

        // column op lit
        let left_input = col("a");
        let left_input = logical2physical(&left_input, &schema);
        let right_input = lit(ScalarValue::Int32(Some(12)));
        let right_input = logical2physical(&right_input, &schema);
        let (result_left, _, result_right) = rewrite_expr_to_prunable(
            &left_input,
            Operator::Eq,
            &right_input,
            df_schema.clone(),
        )
        .unwrap();
        assert_eq!(result_left.to_string(), left_input.to_string());
        assert_eq!(result_right.to_string(), right_input.to_string());

        // cast op lit
        let left_input = cast(col("a"), DataType::Decimal128(20, 3));
        let left_input = logical2physical(&left_input, &schema);
        let right_input = lit(ScalarValue::Decimal128(Some(12), 20, 3));
        let right_input = logical2physical(&right_input, &schema);
        let (result_left, _, result_right) = rewrite_expr_to_prunable(
            &left_input,
            Operator::Gt,
            &right_input,
            df_schema.clone(),
        )
        .unwrap();
        assert_eq!(result_left.to_string(), left_input.to_string());
        assert_eq!(result_right.to_string(), right_input.to_string());

        // try_cast op lit
        let left_input = try_cast(col("a"), DataType::Int64);
        let left_input = logical2physical(&left_input, &schema);
        let right_input = lit(ScalarValue::Int64(Some(12)));
        let right_input = logical2physical(&right_input, &schema);
        let (result_left, _, result_right) =
            rewrite_expr_to_prunable(&left_input, Operator::Gt, &right_input, df_schema)
                .unwrap();
        assert_eq!(result_left.to_string(), left_input.to_string());
        assert_eq!(result_right.to_string(), right_input.to_string());

        // TODO: add test for other case and op
    }

    #[test]
    fn test_rewrite_expr_to_prunable_custom_unhandled_hook() {
        struct CustomUnhandledHook;

        impl UnhandledPredicateHook for CustomUnhandledHook {
            /// This handles an arbitrary case of a column that doesn't exist in the schema
            /// by renaming it to yet another column that doesn't exist in the schema
            /// (the transformation is arbitrary, the point is that it can do whatever it wants)
            fn handle(&self, _expr: &Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {
                Arc::new(phys_expr::Literal::new(ScalarValue::Int32(Some(42))))
            }
        }

        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
        let schema_with_b = Schema::new(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Int32, true),
        ]);

        let rewriter = PredicateRewriter::new()
            .with_unhandled_hook(Arc::new(CustomUnhandledHook {}));

        let transform_expr = |expr| {
            let expr = logical2physical(&expr, &schema_with_b);
            rewriter.rewrite_predicate_to_statistics_predicate(&expr, &schema)
        };

        // transform an arbitrary valid expression that we know is handled
        let known_expression = col("a").eq(lit(12));
        let known_expression_transformed = PredicateRewriter::new()
            .rewrite_predicate_to_statistics_predicate(
                &logical2physical(&known_expression, &schema),
                &schema,
            );

        // an expression referencing an unknown column (that is not in the schema) gets passed to the hook
        let input = col("b").eq(lit(12));
        let expected = logical2physical(&lit(42), &schema);
        let transformed = transform_expr(input.clone());
        assert_eq!(transformed.to_string(), expected.to_string());

        // more complex case with unknown column
        let input = known_expression.clone().and(input.clone());
        let expected = phys_expr::BinaryExpr::new(
            Arc::<dyn PhysicalExpr>::clone(&known_expression_transformed),
            Operator::And,
            logical2physical(&lit(42), &schema),
        );
        let transformed = transform_expr(input.clone());
        assert_eq!(transformed.to_string(), expected.to_string());

        // an unknown expression gets passed to the hook
        let input = array_has(make_array(vec![lit(1)]), col("a"));
        let expected = logical2physical(&lit(42), &schema);
        let transformed = transform_expr(input.clone());
        assert_eq!(transformed.to_string(), expected.to_string());

        // more complex case with unknown expression
        let input = known_expression.and(input);
        let expected = phys_expr::BinaryExpr::new(
            Arc::<dyn PhysicalExpr>::clone(&known_expression_transformed),
            Operator::And,
            logical2physical(&lit(42), &schema),
        );
        let transformed = transform_expr(input.clone());
        assert_eq!(transformed.to_string(), expected.to_string());
    }

    #[test]
    fn test_rewrite_expr_to_prunable_error() {
        // cast string value to numeric value
        // this cast is not supported
        let schema = Schema::new(vec![Field::new("a", DataType::Utf8, true)]);
        let df_schema = DFSchema::try_from(schema.clone()).unwrap();
        let left_input = cast(col("a"), DataType::Int64);
        let left_input = logical2physical(&left_input, &schema);
        let right_input = lit(ScalarValue::Int64(Some(12)));
        let right_input = logical2physical(&right_input, &schema);
        let result = rewrite_expr_to_prunable(
            &left_input,
            Operator::Gt,
            &right_input,
            df_schema.clone(),
        );
        assert!(result.is_err());

        // other expr
        let left_input = is_null(col("a"));
        let left_input = logical2physical(&left_input, &schema);
        let right_input = lit(ScalarValue::Int64(Some(12)));
        let right_input = logical2physical(&right_input, &schema);
        let result =
            rewrite_expr_to_prunable(&left_input, Operator::Gt, &right_input, df_schema);
        assert!(result.is_err());
        // TODO: add other negative test for other case and op
    }

    #[test]
    fn prune_with_contained_one_column() {
        let schema = Arc::new(Schema::new(vec![Field::new("s1", DataType::Utf8, true)]));

        // Model having information like a bloom filter for s1
        let statistics = TestStatistics::new()
            .with_contained(
                "s1",
                [ScalarValue::from("foo")],
                [
                    // container 0 known to only contain "foo"",
                    Some(true),
                    // container 1 known to not contain "foo"
                    Some(false),
                    // container 2 unknown about "foo"
                    None,
                    // container 3 known to only contain "foo"
                    Some(true),
                    // container 4 known to not contain "foo"
                    Some(false),
                    // container 5 unknown about "foo"
                    None,
                    // container 6 known to only contain "foo"
                    Some(true),
                    // container 7 known to not contain "foo"
                    Some(false),
                    // container 8 unknown about "foo"
                    None,
                ],
            )
            .with_contained(
                "s1",
                [ScalarValue::from("bar")],
                [
                    // containers 0,1,2 known to only contain "bar"
                    Some(true),
                    Some(true),
                    Some(true),
                    // container 3,4,5 known to not contain "bar"
                    Some(false),
                    Some(false),
                    Some(false),
                    // container 6,7,8 unknown about "bar"
                    None,
                    None,
                    None,
                ],
            )
            .with_contained(
                // the way the tests are setup, this data is
                // consulted if the "foo" and "bar" are being checked at the same time
                "s1",
                [ScalarValue::from("foo"), ScalarValue::from("bar")],
                [
                    // container 0,1,2 unknown about ("foo, "bar")
                    None,
                    None,
                    None,
                    // container 3,4,5 known to contain only either "foo" and "bar"
                    Some(true),
                    Some(true),
                    Some(true),
                    // container 6,7,8  known to contain  neither "foo" and "bar"
                    Some(false),
                    Some(false),
                    Some(false),
                ],
            );

        // s1 = 'foo'
        prune_with_expr(
            col("s1").eq(lit("foo")),
            &schema,
            &statistics,
            // rule out containers ('false) where we know foo is not present
            &[true, false, true, true, false, true, true, false, true],
        );

        // s1 = 'bar'
        prune_with_expr(
            col("s1").eq(lit("bar")),
            &schema,
            &statistics,
            // rule out containers where we know bar is not present
            &[true, true, true, false, false, false, true, true, true],
        );

        // s1 = 'baz' (unknown value)
        prune_with_expr(
            col("s1").eq(lit("baz")),
            &schema,
            &statistics,
            // can't rule out anything
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 = 'foo' AND s1 = 'bar'
        prune_with_expr(
            col("s1").eq(lit("foo")).and(col("s1").eq(lit("bar"))),
            &schema,
            &statistics,
            // logically this predicate can't possibly be true (the column can't
            // take on both values) but we could rule it out if the stats tell
            // us that both values are not present
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 = 'foo' OR s1 = 'bar'
        prune_with_expr(
            col("s1").eq(lit("foo")).or(col("s1").eq(lit("bar"))),
            &schema,
            &statistics,
            // can rule out containers that we know contain neither foo nor bar
            &[true, true, true, true, true, true, false, false, false],
        );

        // s1 = 'foo' OR s1 = 'baz'
        prune_with_expr(
            col("s1").eq(lit("foo")).or(col("s1").eq(lit("baz"))),
            &schema,
            &statistics,
            // can't rule out anything container
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 = 'foo' OR s1 = 'bar' OR s1 = 'baz'
        prune_with_expr(
            col("s1")
                .eq(lit("foo"))
                .or(col("s1").eq(lit("bar")))
                .or(col("s1").eq(lit("baz"))),
            &schema,
            &statistics,
            // can rule out any containers based on knowledge of s1 and `foo`,
            // `bar` and (`foo`, `bar`)
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 != foo
        prune_with_expr(
            col("s1").not_eq(lit("foo")),
            &schema,
            &statistics,
            // rule out containers we know for sure only contain foo
            &[false, true, true, false, true, true, false, true, true],
        );

        // s1 != bar
        prune_with_expr(
            col("s1").not_eq(lit("bar")),
            &schema,
            &statistics,
            // rule out when we know for sure s1 has the value bar
            &[false, false, false, true, true, true, true, true, true],
        );

        // s1 != foo AND s1 != bar
        prune_with_expr(
            col("s1")
                .not_eq(lit("foo"))
                .and(col("s1").not_eq(lit("bar"))),
            &schema,
            &statistics,
            // can rule out any container where we know s1 does not have either 'foo' or 'bar'
            &[true, true, true, false, false, false, true, true, true],
        );

        // s1 != foo AND s1 != bar AND s1 != baz
        prune_with_expr(
            col("s1")
                .not_eq(lit("foo"))
                .and(col("s1").not_eq(lit("bar")))
                .and(col("s1").not_eq(lit("baz"))),
            &schema,
            &statistics,
            // can't rule out any container based on  knowledge of s1,s2
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 != foo OR s1 != bar
        prune_with_expr(
            col("s1")
                .not_eq(lit("foo"))
                .or(col("s1").not_eq(lit("bar"))),
            &schema,
            &statistics,
            // cant' rule out anything based on contains information
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 != foo OR s1 != bar OR s1 != baz
        prune_with_expr(
            col("s1")
                .not_eq(lit("foo"))
                .or(col("s1").not_eq(lit("bar")))
                .or(col("s1").not_eq(lit("baz"))),
            &schema,
            &statistics,
            // cant' rule out anything based on contains information
            &[true, true, true, true, true, true, true, true, true],
        );
    }

    #[test]
    fn prune_with_contained_two_columns() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("s1", DataType::Utf8, true),
            Field::new("s2", DataType::Utf8, true),
        ]));

        // Model having information like bloom filters for s1 and s2
        let statistics = TestStatistics::new()
            .with_contained(
                "s1",
                [ScalarValue::from("foo")],
                [
                    // container 0, s1 known to only contain "foo"",
                    Some(true),
                    // container 1, s1 known to not contain "foo"
                    Some(false),
                    // container 2, s1 unknown about "foo"
                    None,
                    // container 3, s1 known to only contain "foo"
                    Some(true),
                    // container 4, s1 known to not contain "foo"
                    Some(false),
                    // container 5, s1 unknown about "foo"
                    None,
                    // container 6, s1 known to only contain "foo"
                    Some(true),
                    // container 7, s1 known to not contain "foo"
                    Some(false),
                    // container 8, s1 unknown about "foo"
                    None,
                ],
            )
            .with_contained(
                "s2", // for column s2
                [ScalarValue::from("bar")],
                [
                    // containers 0,1,2 s2 known to only contain "bar"
                    Some(true),
                    Some(true),
                    Some(true),
                    // container 3,4,5 s2 known to not contain "bar"
                    Some(false),
                    Some(false),
                    Some(false),
                    // container 6,7,8 s2 unknown about "bar"
                    None,
                    None,
                    None,
                ],
            );

        // s1 = 'foo'
        prune_with_expr(
            col("s1").eq(lit("foo")),
            &schema,
            &statistics,
            // rule out containers where we know s1 is not present
            &[true, false, true, true, false, true, true, false, true],
        );

        // s1 = 'foo' OR s2 = 'bar'
        let expr = col("s1").eq(lit("foo")).or(col("s2").eq(lit("bar")));
        prune_with_expr(
            expr,
            &schema,
            &statistics,
            //  can't rule out any container (would need to prove that s1 != foo AND s2 != bar)
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 = 'foo' AND s2 != 'bar'
        prune_with_expr(
            col("s1").eq(lit("foo")).and(col("s2").not_eq(lit("bar"))),
            &schema,
            &statistics,
            // can only rule out container where we know either:
            // 1. s1 doesn't have the value 'foo` or
            // 2. s2 has only the value of 'bar'
            &[false, false, false, true, false, true, true, false, true],
        );

        // s1 != 'foo' AND s2 != 'bar'
        prune_with_expr(
            col("s1")
                .not_eq(lit("foo"))
                .and(col("s2").not_eq(lit("bar"))),
            &schema,
            &statistics,
            // Can  rule out any container where we know either
            // 1. s1 has only the value 'foo'
            // 2. s2 has only the value 'bar'
            &[false, false, false, false, true, true, false, true, true],
        );

        // s1 != 'foo' AND (s2 = 'bar' OR s2 = 'baz')
        prune_with_expr(
            col("s1")
                .not_eq(lit("foo"))
                .and(col("s2").eq(lit("bar")).or(col("s2").eq(lit("baz")))),
            &schema,
            &statistics,
            // Can rule out any container where we know s1 has only the value
            // 'foo'. Can't use knowledge of s2 and bar to rule out anything
            &[false, true, true, false, true, true, false, true, true],
        );

        // s1 like '%foo%bar%'
        prune_with_expr(
            col("s1").like(lit("foo%bar%")),
            &schema,
            &statistics,
            // cant rule out anything with information we know
            &[true, true, true, true, true, true, true, true, true],
        );

        // s1 like '%foo%bar%' AND s2 = 'bar'
        prune_with_expr(
            col("s1")
                .like(lit("foo%bar%"))
                .and(col("s2").eq(lit("bar"))),
            &schema,
            &statistics,
            // can rule out any container where we know s2 does not have the value 'bar'
            &[true, true, true, false, false, false, true, true, true],
        );

        // s1 like '%foo%bar%' OR s2 = 'bar'
        prune_with_expr(
            col("s1").like(lit("foo%bar%")).or(col("s2").eq(lit("bar"))),
            &schema,
            &statistics,
            // can't rule out anything (we would have to prove that both the
            // like and the equality must be false)
            &[true, true, true, true, true, true, true, true, true],
        );
    }

    #[test]
    fn prune_with_range_and_contained() {
        // Setup mimics range information for i, a bloom filter for s
        let schema = Arc::new(Schema::new(vec![
            Field::new("i", DataType::Int32, true),
            Field::new("s", DataType::Utf8, true),
        ]));

        let statistics = TestStatistics::new()
            .with(
                "i",
                ContainerStats::new_i32(
                    // Container 0, 3, 6: [-5 to 5]
                    // Container 1, 4, 7: [10 to 20]
                    // Container 2, 5, 9: unknown
                    vec![
                        Some(-5),
                        Some(10),
                        None,
                        Some(-5),
                        Some(10),
                        None,
                        Some(-5),
                        Some(10),
                        None,
                    ], // min
                    vec![
                        Some(5),
                        Some(20),
                        None,
                        Some(5),
                        Some(20),
                        None,
                        Some(5),
                        Some(20),
                        None,
                    ], // max
                ),
            )
            // Add contained  information about the s and "foo"
            .with_contained(
                "s",
                [ScalarValue::from("foo")],
                [
                    // container 0,1,2 known to only contain "foo"
                    Some(true),
                    Some(true),
                    Some(true),
                    // container 3,4,5 known to not contain "foo"
                    Some(false),
                    Some(false),
                    Some(false),
                    // container 6,7,8 unknown about "foo"
                    None,
                    None,
                    None,
                ],
            );

        // i = 0 and s = 'foo'
        prune_with_expr(
            col("i").eq(lit(0)).and(col("s").eq(lit("foo"))),
            &schema,
            &statistics,
            // Can rule out container where we know that either:
            // 1. 0 is outside the min/max range of i
            // 1. s does not contain foo
            // (range is false, and contained  is false)
            &[true, false, true, false, false, false, true, false, true],
        );

        // i = 0 and s != 'foo'
        prune_with_expr(
            col("i").eq(lit(0)).and(col("s").not_eq(lit("foo"))),
            &schema,
            &statistics,
            // Can rule out containers where either:
            // 1. 0 is outside the min/max range of i
            // 2. s only contains foo
            &[false, false, false, true, false, true, true, false, true],
        );

        // i = 0 OR s = 'foo'
        prune_with_expr(
            col("i").eq(lit(0)).or(col("s").eq(lit("foo"))),
            &schema,
            &statistics,
            // in theory could rule out containers if we had min/max values for
            // s as well. But in this case we don't so we can't rule out anything
            &[true, true, true, true, true, true, true, true, true],
        );
    }

    /// prunes the specified expr with the specified schema and statistics, and
    /// ensures it returns expected.
    ///
    /// `expected` is a vector of bools, where true means the row group should
    /// be kept, and false means it should be pruned.
    // TODO refactor other tests to use this to reduce boiler plate
    fn prune_with_expr(
        expr: Expr,
        schema: &SchemaRef,
        statistics: &TestStatistics,
        expected: &[bool],
    ) {
        println!("Pruning with expr: {expr}");
        let expr = logical2physical(&expr, schema);
        let p = PruningPredicate::try_new(expr, Arc::<Schema>::clone(schema)).unwrap();
        let result = p.prune(statistics).unwrap();
        assert_eq!(result, expected);
    }

    fn prune_with_simplified_expr(
        expr: Expr,
        schema: &SchemaRef,
        statistics: &TestStatistics,
        expected: &[bool],
    ) {
        println!("Pruning with expr: {expr}");
        let expr = logical2physical(&expr, schema);
        let simplifier = PhysicalExprSimplifier::new(schema);
        let expr = simplifier.simplify(expr).unwrap();
        let p = PruningPredicate::try_new(expr, Arc::<Schema>::clone(schema)).unwrap();
        let result = p.prune(statistics).unwrap();
        assert_eq!(result, expected);
    }

    fn is_not_distinct_from(left: Expr, right: Expr) -> Expr {
        Expr::BinaryExpr(BinaryExpr::new(
            Box::new(left),
            Operator::IsNotDistinctFrom,
            Box::new(right),
        ))
    }

    fn is_distinct_from(left: Expr, right: Expr) -> Expr {
        Expr::BinaryExpr(BinaryExpr::new(
            Box::new(left),
            Operator::IsDistinctFrom,
            Box::new(right),
        ))
    }

    fn test_build_predicate_expression(
        expr: &Expr,
        schema: &Schema,
        required_columns: &mut RequiredColumns,
    ) -> Arc<dyn PhysicalExpr> {
        let expr = logical2physical(expr, schema);
        let unhandled_hook = Arc::new(ConstantUnhandledPredicateHook::default()) as _;
        build_predicate_expression(
            &expr,
            &Arc::new(schema.clone()),
            required_columns,
            &unhandled_hook,
        )
    }

    #[test]
    fn test_build_predicate_expression_with_false() {
        let expr = lit(ScalarValue::Boolean(Some(false)));
        let schema = Schema::empty();
        let res =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        let expected = logical2physical(&expr, &schema);
        assert_eq!(&res, &expected);
    }

    #[test]
    fn test_build_predicate_expression_with_and_false() {
        let schema = Schema::new(vec![Field::new("c1", DataType::Utf8View, false)]);
        let expr = and(
            col("c1").eq(lit("a")),
            lit(ScalarValue::Boolean(Some(false))),
        );
        let res =
            test_build_predicate_expression(&expr, &schema, &mut RequiredColumns::new());
        let expected = logical2physical(&lit(ScalarValue::Boolean(Some(false))), &schema);
        assert_eq!(&res, &expected);
    }

    #[test]
    fn test_build_predicate_expression_with_or_false() {
        let schema = Schema::new(vec![Field::new("c1", DataType::Utf8View, false)]);
        let left_expr = col("c1").eq(lit("a"));
        let right_expr = lit(ScalarValue::Boolean(Some(false)));
        let res = test_build_predicate_expression(
            &or(left_expr.clone(), right_expr.clone()),
            &schema,
            &mut RequiredColumns::new(),
        );
        let expected =
            "c1_null_count@2 != row_count@3 AND c1_min@0 <= a AND a <= c1_max@1";
        assert_eq!(res.to_string(), expected);
    }
}