datafusion-physical-expr 55.0.0

// 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.

mod kernels;

use crate::PhysicalExpr;
use crate::intervals::cp_solver::{propagate_arithmetic, propagate_comparison};
use std::cmp::Ordering;
use std::hash::Hash;
use std::sync::Arc;

use arrow::array::*;
use arrow::compute::kernels::boolean::{and_kleene, or_kleene};
use arrow::compute::kernels::concat_elements::concat_elements_dyn;
use arrow::compute::{SlicesIterator, cast, filter_record_batch};
use arrow::datatypes::*;
use arrow::error::ArrowError;
use datafusion_common::cast::as_boolean_array;
use datafusion_common::{Result, ScalarValue, internal_err, not_impl_err};

use datafusion_expr::binary::BinaryTypeCoercer;
use datafusion_expr::interval_arithmetic::{Interval, apply_operator};
use datafusion_expr::sort_properties::{ExprProperties, SortProperties};
#[expect(deprecated)]
use datafusion_expr::statistics::Distribution::{Bernoulli, Gaussian};
#[expect(deprecated)]
use datafusion_expr::statistics::{
    Distribution, combine_bernoullis, combine_gaussians,
    create_bernoulli_from_comparison, new_generic_from_binary_op,
};
use datafusion_expr::{ColumnarValue, Operator};
use datafusion_physical_expr_common::datum::{apply, apply_cmp};

use kernels::{
    bitwise_and_dyn, bitwise_and_dyn_scalar, bitwise_or_dyn, bitwise_or_dyn_scalar,
    bitwise_shift_left_dyn, bitwise_shift_left_dyn_scalar, bitwise_shift_right_dyn,
    bitwise_shift_right_dyn_scalar, bitwise_xor_dyn, bitwise_xor_dyn_scalar,
    regex_match_dyn, regex_match_dyn_scalar,
};

/// Binary expression
#[derive(Debug, Clone, Eq)]
pub struct BinaryExpr {
    left: Arc<dyn PhysicalExpr>,
    op: Operator,
    right: Arc<dyn PhysicalExpr>,
    /// Specifies whether an error is returned on overflow or not
    fail_on_overflow: bool,
}

// Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808
impl PartialEq for BinaryExpr {
    fn eq(&self, other: &Self) -> bool {
        self.left.eq(&other.left)
            && self.op.eq(&other.op)
            && self.right.eq(&other.right)
            && self.fail_on_overflow.eq(&other.fail_on_overflow)
    }
}
impl Hash for BinaryExpr {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.left.hash(state);
        self.op.hash(state);
        self.right.hash(state);
        self.fail_on_overflow.hash(state);
    }
}

impl BinaryExpr {
    /// Create new binary expression
    pub fn new(
        left: Arc<dyn PhysicalExpr>,
        op: Operator,
        right: Arc<dyn PhysicalExpr>,
    ) -> Self {
        Self {
            left,
            op,
            right,
            fail_on_overflow: false,
        }
    }

    /// Create new binary expression with explicit fail_on_overflow value
    pub fn with_fail_on_overflow(self, fail_on_overflow: bool) -> Self {
        Self {
            left: self.left,
            op: self.op,
            right: self.right,
            fail_on_overflow,
        }
    }

    /// Get the left side of the binary expression
    pub fn left(&self) -> &Arc<dyn PhysicalExpr> {
        &self.left
    }

    /// Get the right side of the binary expression
    pub fn right(&self) -> &Arc<dyn PhysicalExpr> {
        &self.right
    }

    /// Get the operator for this binary expression
    pub fn op(&self) -> &Operator {
        &self.op
    }

    /// Wrapping on overflow breaks monotonicity (e.g. the sum of two
    /// ascending `UInt8` columns can wrap back to small values), so the
    /// derived ordering is kept only when overflow is impossible. `time ±
    /// interval` wraps around the 24-hour clock even in checked mode, so it
    /// never preserves ordering.
    fn arithmetic_sort_properties(
        &self,
        sort_properties: SortProperties,
        l_range: &Interval,
        r_range: &Interval,
        range: &Interval,
    ) -> SortProperties {
        if sort_properties == SortProperties::Singleton {
            return sort_properties;
        }
        let wraps_in_domain = match self.op {
            Operator::Plus => {
                is_time_plus_interval(&l_range.data_type(), &r_range.data_type())
            }
            Operator::Minus => {
                is_time_minus_interval(&l_range.data_type(), &r_range.data_type())
            }
            _ => false,
        };
        let cannot_overflow = !range.is_unbounded()
            && !unsigned_subtraction_may_underflow(self.op, l_range, r_range, range);
        if !wraps_in_domain && (self.fail_on_overflow || cannot_overflow) {
            sort_properties
        } else {
            SortProperties::Unordered
        }
    }
}

/// Returns `true` unless `l_range - r_range` provably stays within an unsigned
/// domain.
///
/// [`Interval`] standardizes an underflowed (i.e. `null`) lower bound of an
/// unsigned type back to zero, so an apparently bounded result range is not
/// enough to rule out wrapping here -- e.g. `[0, 10] - [0, 10]` over `UInt32`
/// yields `[0, 10]` even though `0 - 10` wraps to `u32::MAX`. Compare the
/// endpoints that produce the smallest difference instead.
fn unsigned_subtraction_may_underflow(
    op: Operator,
    l_range: &Interval,
    r_range: &Interval,
    range: &Interval,
) -> bool {
    if op != Operator::Minus || !range.data_type().is_unsigned_integer() {
        return false;
    }
    let (smallest_lhs, largest_rhs) = (l_range.lower(), r_range.upper());
    if smallest_lhs.is_null() || largest_rhs.is_null() {
        return true;
    }
    // Operands of differing types compare as incomparable, in which case we
    // conservatively assume an underflow is possible.
    !matches!(
        smallest_lhs.partial_cmp(largest_rhs),
        Some(Ordering::Greater | Ordering::Equal)
    )
}

impl std::fmt::Display for BinaryExpr {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        // Put parentheses around child binary expressions so that we can see the difference
        // between `(a OR b) AND c` and `a OR (b AND c)`. We only insert parentheses when needed,
        // based on operator precedence. For example, `(a AND b) OR c` and `a AND b OR c` are
        // equivalent and the parentheses are not necessary.

        fn write_child(
            f: &mut std::fmt::Formatter,
            expr: &dyn PhysicalExpr,
            precedence: u8,
        ) -> std::fmt::Result {
            if let Some(child) = expr.downcast_ref::<BinaryExpr>() {
                let p = child.op.precedence();
                if p == 0 || p < precedence {
                    write!(f, "({child})")?;
                } else {
                    write!(f, "{child}")?;
                }
            } else {
                write!(f, "{expr}")?;
            }

            Ok(())
        }

        let precedence = self.op.precedence();
        write_child(f, self.left.as_ref(), precedence)?;
        write!(f, " {} ", self.op)?;
        write_child(f, self.right.as_ref(), precedence)
    }
}

/// Invoke a boolean kernel on a pair of arrays
#[inline]
fn boolean_op(
    left: &dyn Array,
    right: &dyn Array,
    op: impl FnOnce(&BooleanArray, &BooleanArray) -> Result<BooleanArray, ArrowError>,
) -> Result<Arc<dyn Array + 'static>, ArrowError> {
    let ll = as_boolean_array(left).expect("boolean_op failed to downcast left array");
    let rr = as_boolean_array(right).expect("boolean_op failed to downcast right array");
    op(ll, rr).map(|t| Arc::new(t) as _)
}

/// Returns true if both operands are Date types (Date32 or Date64)
/// Used to detect Date - Date operations which should return Int64 (days difference)
fn is_date_minus_date(lhs: &DataType, rhs: &DataType) -> bool {
    matches!(
        (lhs, rhs),
        (DataType::Date32, DataType::Date32) | (DataType::Date64, DataType::Date64)
    )
}

/// Milliseconds per day, used for Date64 subtraction.
const MILLIS_PER_DAY: i64 = 86_400_000;

/// Evaluates `Date32 - Date32` or `Date64 - Date64`, returning the difference in
/// whole days as `Int64`.
///
/// This matches the behavior of PostgreSQL, DuckDB, and MySQL, where
/// `date - date` yields an integer day count rather than an interval.
fn apply_date_subtraction(
    lhs: &ColumnarValue,
    rhs: &ColumnarValue,
) -> Result<ColumnarValue> {
    match (lhs.data_type(), rhs.data_type()) {
        (DataType::Date32, DataType::Date32) => {
            subtract_date_to_days::<Date32Type>(lhs, rhs, |l, r| l - r)
        }
        (DataType::Date64, DataType::Date64) => {
            subtract_date_to_days::<Date64Type>(lhs, rhs, |l, r| {
                l.wrapping_sub(r) / MILLIS_PER_DAY
            })
        }
        (_, _) => unreachable!("apply_date_subtraction called with non-date types"),
    }
}

/// Generic date subtraction: operates directly on the native primitive values
/// of `T` (i32 for Date32, i64 for Date64), applying `day_diff_fn` to produce
/// an Int64 day count.
fn subtract_date_to_days<T: ArrowPrimitiveType>(
    lhs: &ColumnarValue,
    rhs: &ColumnarValue,
    day_diff_fn: impl Fn(i64, i64) -> i64,
) -> Result<ColumnarValue>
where
    T::Native: Copy + Into<i64>,
{
    /// Extract the date value as `i64`. Returns `None` for null scalars.
    fn date_scalar_to_i64<P: ArrowPrimitiveType>(
        scalar: &ScalarValue,
    ) -> Result<Option<i64>> {
        match scalar {
            ScalarValue::Date32(value) if P::DATA_TYPE == DataType::Date32 => {
                Ok(value.map(i64::from))
            }
            ScalarValue::Date64(value) if P::DATA_TYPE == DataType::Date64 => Ok(*value),
            other => {
                internal_err!(
                    "{} date scalar expected, got: {}",
                    P::DATA_TYPE,
                    other.data_type()
                )
            }
        }
    }

    match (lhs, rhs) {
        (ColumnarValue::Array(left), ColumnarValue::Array(right)) => {
            let left = left.as_primitive::<T>();
            let right = right.as_primitive::<T>();
            let result: Int64Array =
                arrow::compute::binary::<_, _, _, Int64Type>(left, right, |l, r| {
                    day_diff_fn(l.into(), r.into())
                })?;
            Ok(ColumnarValue::Array(Arc::new(result)))
        }
        (ColumnarValue::Array(left), ColumnarValue::Scalar(right)) => {
            let left = left.as_primitive::<T>();
            match date_scalar_to_i64::<T>(right)? {
                Some(right_val) => {
                    let result: Int64Array =
                        left.unary(|l| day_diff_fn(l.into(), right_val));
                    Ok(ColumnarValue::Array(Arc::new(result)))
                }
                None => Ok(ColumnarValue::Scalar(ScalarValue::Int64(None))),
            }
        }
        (ColumnarValue::Scalar(left), ColumnarValue::Array(right)) => {
            let right = right.as_primitive::<T>();
            match date_scalar_to_i64::<T>(left)? {
                Some(left_val) => {
                    let result: Int64Array =
                        right.unary(|r| day_diff_fn(left_val, r.into()));
                    Ok(ColumnarValue::Array(Arc::new(result)))
                }
                None => Ok(ColumnarValue::Scalar(ScalarValue::Int64(None))),
            }
        }
        (ColumnarValue::Scalar(left), ColumnarValue::Scalar(right)) => {
            let left_val = date_scalar_to_i64::<T>(left)?;
            let right_val = date_scalar_to_i64::<T>(right)?;
            Ok(ColumnarValue::Scalar(ScalarValue::Int64(
                left_val.zip(right_val).map(|(l, r)| day_diff_fn(l, r)),
            )))
        }
    }
}

/// Returns true for `time + interval` or `interval + time`.
fn is_time_plus_interval(lhs: &DataType, rhs: &DataType) -> bool {
    matches!(
        (lhs, rhs),
        (
            DataType::Time32(_) | DataType::Time64(_),
            DataType::Interval(_)
        ) | (
            DataType::Interval(_),
            DataType::Time32(_) | DataType::Time64(_)
        )
    )
}

/// Returns true for `time - interval`.
fn is_time_minus_interval(lhs: &DataType, rhs: &DataType) -> bool {
    matches!(
        (lhs, rhs),
        (
            DataType::Time32(_) | DataType::Time64(_),
            DataType::Interval(_)
        )
    )
}

/// Evaluates `time + interval`, `interval + time`, or `time - interval`, returning a
/// `time` wrapped within the 24-hour clock to match PostgreSQL and DuckDB (e.g.
/// `time '23:30' + interval '2 hours'` is `01:30:00`). arrow's arithmetic kernels do
/// not implement time-of-day arithmetic, so it is handled here.
///
/// The result keeps the input time's unit; the interval (normalized to `MonthDayNano`
/// by the coercion layer) is applied at nanosecond precision and floored to that unit,
/// mirroring `timestamp(unit) + interval`. Only the sub-day portion of the interval
/// affects a time-of-day -- whole months and days are ignored, matching PostgreSQL. The
/// floor is applied after the sign, so `time(s) + interval '1 nanosecond'` is a no-op
/// while `time(s) - interval '1 nanosecond'` rolls back a second, exactly as the
/// timestamp case does.
fn apply_time_interval(
    lhs: &ColumnarValue,
    rhs: &ColumnarValue,
    subtract: bool,
) -> Result<ColumnarValue> {
    // The `time` operand determines the result type; the other is the interval.
    let (time, interval) = if matches!(lhs.data_type(), DataType::Interval(_)) {
        (rhs, lhs)
    } else {
        (lhs, rhs)
    };

    // Dispatch on the time unit; `ns_per_unit` converts the interval's nanoseconds to
    // that unit, and the arithmetic is done (and wrapped) at that resolution.
    match time.data_type() {
        DataType::Time32(TimeUnit::Second) => wrap_time_interval::<Time32SecondType>(
            time,
            interval,
            subtract,
            1_000_000_000,
        ),
        DataType::Time32(TimeUnit::Millisecond) => {
            wrap_time_interval::<Time32MillisecondType>(
                time, interval, subtract, 1_000_000,
            )
        }
        DataType::Time64(TimeUnit::Microsecond) => {
            wrap_time_interval::<Time64MicrosecondType>(time, interval, subtract, 1_000)
        }
        DataType::Time64(TimeUnit::Nanosecond) => {
            wrap_time_interval::<Time64NanosecondType>(time, interval, subtract, 1)
        }
        other => internal_err!("time operand expected, got: {other}"),
    }
}

/// Adds or subtracts an interval to/from a `time` of arrow primitive type `T`, wrapping
/// the result within the 24-hour clock and keeping the type `T`. `ns_per_unit` is the
/// number of nanoseconds in one unit of `T` (e.g. `1_000` for microseconds).
fn wrap_time_interval<T: ArrowPrimitiveType>(
    time: &ColumnarValue,
    interval: &ColumnarValue,
    subtract: bool,
    ns_per_unit: i64,
) -> Result<ColumnarValue>
where
    T::Native: Copy + Into<i64> + TryFrom<i64>,
{
    /// Nanoseconds in a 24-hour day.
    const DAY_NANOS: i64 = 86_400_000_000_000;
    // Units in a 24-hour day, at `T`'s resolution.
    let day_units = DAY_NANOS / ns_per_unit;

    // Wraps `time ± interval` into `[0, day_units)`. The interval is reduced modulo a day
    // (so the sum stays within `i64`), applied at nanosecond precision, then floored to
    // `T`'s unit -- matching `timestamp(unit) ± interval`. Because the floor is applied
    // after the sign, `time(s) - interval '1 nanosecond'` rolls back a full second, just
    // as the timestamp case does, while `time(s) + interval '1 nanosecond'` is a no-op.
    // `div_euclid`/`rem_euclid` floor toward negative infinity, so the wrapped value stays
    // in `[0, day_units)`, which always fits `T::Native`.
    let wrap = |time_unit: i64, iv: IntervalMonthDayNano| -> T::Native {
        let iv_ns = iv.nanoseconds % DAY_NANOS;
        let signed_ns = if subtract { -iv_ns } else { iv_ns };
        let delta = signed_ns.div_euclid(ns_per_unit);
        let wrapped = (time_unit + delta).rem_euclid(day_units);
        T::Native::try_from(wrapped).unwrap_or_default()
    };

    /// Extracts an `Interval(MonthDayNano)` scalar.
    fn interval_scalar(scalar: &ScalarValue) -> Result<Option<IntervalMonthDayNano>> {
        match scalar {
            ScalarValue::IntervalMonthDayNano(value) => Ok(*value),
            other => internal_err!(
                "Interval(MonthDayNano) scalar expected, got: {}",
                other.data_type()
            ),
        }
    }

    /// Extracts a time scalar as its unit count since midnight.
    fn time_scalar_units(scalar: &ScalarValue) -> Result<Option<i64>> {
        match scalar {
            ScalarValue::Time32Second(value) | ScalarValue::Time32Millisecond(value) => {
                Ok(value.map(i64::from))
            }
            ScalarValue::Time64Microsecond(value)
            | ScalarValue::Time64Nanosecond(value) => Ok(*value),
            other => {
                internal_err!("time scalar expected, got: {}", other.data_type())
            }
        }
    }

    /// Builds a time scalar of type `P` from a unit count.
    fn time_scalar<P: ArrowPrimitiveType>(value: Option<i64>) -> ScalarValue {
        match P::DATA_TYPE {
            DataType::Time32(TimeUnit::Second) => {
                ScalarValue::Time32Second(value.map(|v| v as i32))
            }
            DataType::Time32(TimeUnit::Millisecond) => {
                ScalarValue::Time32Millisecond(value.map(|v| v as i32))
            }
            DataType::Time64(TimeUnit::Microsecond) => {
                ScalarValue::Time64Microsecond(value)
            }
            _ => ScalarValue::Time64Nanosecond(value),
        }
    }

    match (time, interval) {
        (ColumnarValue::Array(time), ColumnarValue::Array(interval)) => {
            let time = time.as_primitive::<T>();
            let interval = interval.as_primitive::<IntervalMonthDayNanoType>();
            let result: PrimitiveArray<T> =
                arrow::compute::binary(time, interval, |t, iv| wrap(t.into(), iv))?;
            Ok(ColumnarValue::Array(Arc::new(result)))
        }
        (ColumnarValue::Array(time), ColumnarValue::Scalar(interval)) => {
            let time = time.as_primitive::<T>();
            match interval_scalar(interval)? {
                Some(iv) => {
                    let result: PrimitiveArray<T> = time.unary(|t| wrap(t.into(), iv));
                    Ok(ColumnarValue::Array(Arc::new(result)))
                }
                None => Ok(ColumnarValue::Scalar(time_scalar::<T>(None))),
            }
        }
        (ColumnarValue::Scalar(time), ColumnarValue::Array(interval)) => {
            let interval = interval.as_primitive::<IntervalMonthDayNanoType>();
            match time_scalar_units(time)? {
                Some(t) => {
                    let result: PrimitiveArray<T> = interval.unary(|iv| wrap(t, iv));
                    Ok(ColumnarValue::Array(Arc::new(result)))
                }
                None => Ok(ColumnarValue::Scalar(time_scalar::<T>(None))),
            }
        }
        (ColumnarValue::Scalar(time), ColumnarValue::Scalar(interval)) => {
            let result = time_scalar_units(time)?
                .zip(interval_scalar(interval)?)
                .map(|(t, iv)| wrap(t, iv).into());
            Ok(ColumnarValue::Scalar(time_scalar::<T>(result)))
        }
    }
}

impl PhysicalExpr for BinaryExpr {
    fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
        BinaryTypeCoercer::new(
            &self.left.data_type(input_schema)?,
            &self.op,
            &self.right.data_type(input_schema)?,
        )
        .get_result_type()
    }

    fn nullable(&self, input_schema: &Schema) -> Result<bool> {
        Ok(self.left.nullable(input_schema)? || self.right.nullable(input_schema)?)
    }

    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
        use arrow::compute::kernels::numeric::*;

        // Evaluate left-hand side expression.
        let lhs = self.left.evaluate(batch)?;

        // Check if we can apply short-circuit evaluation.
        match check_short_circuit(&lhs, &self.op) {
            ShortCircuitStrategy::None => {}
            ShortCircuitStrategy::ReturnLeft => return Ok(lhs),
            ShortCircuitStrategy::ReturnRight => {
                let rhs = self.right.evaluate(batch)?;
                return Ok(rhs);
            }
            ShortCircuitStrategy::PreSelection { mask, fill_value } => {
                // `mask` selects the rows whose result depends on the RHS; the
                // unselected rows are all `fill_value` (see `ShortCircuitStrategy`).
                //
                // Use `filter_record_batch` directly because `evaluate_selection`
                // scatters the RHS back to the original batch length.
                let selection_batch = filter_record_batch(batch, &mask)?;
                let right_ret = self.right.evaluate(&selection_batch)?;

                match &right_ret {
                    ColumnarValue::Array(array) => {
                        let boolean_array = array.as_boolean();
                        // If the RHS is uniform on the selected rows, the whole
                        // expression collapses and no scatter is needed.
                        if boolean_array.null_count() == 0 {
                            let rhs_value = if !boolean_array.has_false() {
                                Some(true)
                            } else if !boolean_array.has_true() {
                                Some(false)
                            } else {
                                None
                            };
                            if let Some(rhs_value) = rhs_value {
                                return Ok(uniform_pre_selection_result(
                                    rhs_value, fill_value, lhs,
                                ));
                            }
                        }

                        return pre_selection_scatter(
                            &mask,
                            Some(boolean_array),
                            fill_value,
                        );
                    }
                    ColumnarValue::Scalar(scalar) => {
                        if let ScalarValue::Boolean(v) = scalar {
                            // A scalar RHS applies uniformly to all selected rows.
                            if let Some(v) = v {
                                return Ok(uniform_pre_selection_result(
                                    *v, fill_value, lhs,
                                ));
                            } else {
                                return pre_selection_scatter(&mask, None, fill_value);
                            }
                        } else {
                            return internal_err!(
                                "Expected boolean scalar value, found: {right_ret:?}"
                            );
                        }
                    }
                }
            }
        }

        let rhs = self.right.evaluate(batch)?;
        let left_data_type = lhs.data_type();
        let right_data_type = rhs.data_type();

        let schema = batch.schema();
        let input_schema = schema.as_ref();

        match self.op {
            // `time ± interval` returns a wrapped `time` (PostgreSQL/DuckDB
            // semantics); arrow's arithmetic kernels don't implement it.
            Operator::Plus
                if is_time_plus_interval(&left_data_type, &right_data_type) =>
            {
                return apply_time_interval(&lhs, &rhs, false);
            }
            Operator::Minus
                if is_time_minus_interval(&left_data_type, &right_data_type) =>
            {
                return apply_time_interval(&lhs, &rhs, true);
            }
            Operator::Plus if self.fail_on_overflow => return apply(&lhs, &rhs, add),
            Operator::Plus => return apply(&lhs, &rhs, add_wrapping),
            // Special case: Date - Date returns Int64 (days difference)
            // This aligns with PostgreSQL, DuckDB, and MySQL behavior
            Operator::Minus if is_date_minus_date(&left_data_type, &right_data_type) => {
                return apply_date_subtraction(&lhs, &rhs);
            }
            Operator::Minus if self.fail_on_overflow => return apply(&lhs, &rhs, sub),
            Operator::Minus => return apply(&lhs, &rhs, sub_wrapping),
            Operator::Multiply if self.fail_on_overflow => return apply(&lhs, &rhs, mul),
            Operator::Multiply => return apply(&lhs, &rhs, mul_wrapping),
            Operator::Divide => return apply(&lhs, &rhs, div),
            Operator::Modulo => return apply(&lhs, &rhs, rem),

            Operator::Eq
            | Operator::NotEq
            | Operator::Lt
            | Operator::Gt
            | Operator::LtEq
            | Operator::GtEq
            | Operator::IsDistinctFrom
            | Operator::IsNotDistinctFrom
            | Operator::LikeMatch
            | Operator::ILikeMatch
            | Operator::NotLikeMatch
            | Operator::NotILikeMatch => {
                return apply_cmp(self.op, &lhs, &rhs);
            }
            _ => {}
        }

        let result_type = self.data_type(input_schema)?;

        // If the left-hand side is an array and the right-hand side is a non-null scalar, try the optimized kernel.
        if let (ColumnarValue::Array(array), ColumnarValue::Scalar(scalar)) = (&lhs, &rhs)
            && !scalar.is_null()
            && let Some(result_array) =
                self.evaluate_array_scalar(array, scalar.clone())?
        {
            let final_array = result_array
                .and_then(|a| to_result_type_array(&self.op, a, &result_type));
            return final_array.map(ColumnarValue::Array);
        }

        // if both arrays or both literals - extract arrays and continue execution
        let (left, right) = (
            lhs.into_array(batch.num_rows())?,
            rhs.into_array(batch.num_rows())?,
        );
        self.evaluate_with_resolved_args(left, &left_data_type, right, &right_data_type)
            .map(ColumnarValue::Array)
    }

    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
        vec![&self.left, &self.right]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn PhysicalExpr>>,
    ) -> Result<Arc<dyn PhysicalExpr>> {
        Ok(Arc::new(
            BinaryExpr::new(Arc::clone(&children[0]), self.op, Arc::clone(&children[1]))
                .with_fail_on_overflow(self.fail_on_overflow),
        ))
    }

    fn evaluate_bounds(&self, children: &[&Interval]) -> Result<Interval> {
        // Get children intervals:
        let left_interval = children[0];
        let right_interval = children[1];
        // Calculate current node's interval:
        apply_operator(&self.op, left_interval, right_interval)
    }

    fn propagate_constraints(
        &self,
        interval: &Interval,
        children: &[&Interval],
    ) -> Result<Option<Vec<Interval>>> {
        // Get children intervals.
        let left_interval = children[0];
        let right_interval = children[1];

        if self.op.eq(&Operator::And) {
            if interval.eq(&Interval::TRUE) {
                // A certainly true logical conjunction can only derive from possibly
                // true operands. Otherwise, we prove infeasibility.
                Ok((!left_interval.eq(&Interval::FALSE)
                    && !right_interval.eq(&Interval::FALSE))
                .then(|| vec![Interval::TRUE, Interval::TRUE]))
            } else if interval.eq(&Interval::FALSE) {
                // If the logical conjunction is certainly false, one of the
                // operands must be false. However, it's not always possible to
                // determine which operand is false, leading to different scenarios.

                // If one operand is certainly true and the other one is uncertain,
                // then the latter must be certainly false.
                if left_interval.eq(&Interval::TRUE)
                    && right_interval.eq(&Interval::TRUE_OR_FALSE)
                {
                    Ok(Some(vec![Interval::TRUE, Interval::FALSE]))
                } else if right_interval.eq(&Interval::TRUE)
                    && left_interval.eq(&Interval::TRUE_OR_FALSE)
                {
                    Ok(Some(vec![Interval::FALSE, Interval::TRUE]))
                }
                // If both children are uncertain, or if one is certainly false,
                // we cannot conclusively refine their intervals. In this case,
                // propagation does not result in any interval changes.
                else {
                    Ok(Some(vec![]))
                }
            } else {
                // An uncertain logical conjunction result can not shrink the
                // end-points of its children.
                Ok(Some(vec![]))
            }
        } else if self.op.eq(&Operator::Or) {
            if interval.eq(&Interval::FALSE) {
                // A certainly false logical disjunction can only derive from certainly
                // false operands. Otherwise, we prove infeasibility.
                Ok((!left_interval.eq(&Interval::TRUE)
                    && !right_interval.eq(&Interval::TRUE))
                .then(|| vec![Interval::FALSE, Interval::FALSE]))
            } else if interval.eq(&Interval::TRUE) {
                // If the logical disjunction is certainly true, one of the
                // operands must be true. However, it's not always possible to
                // determine which operand is true, leading to different scenarios.

                // If one operand is certainly false and the other one is uncertain,
                // then the latter must be certainly true.
                if left_interval.eq(&Interval::FALSE)
                    && right_interval.eq(&Interval::TRUE_OR_FALSE)
                {
                    Ok(Some(vec![Interval::FALSE, Interval::TRUE]))
                } else if right_interval.eq(&Interval::FALSE)
                    && left_interval.eq(&Interval::TRUE_OR_FALSE)
                {
                    Ok(Some(vec![Interval::TRUE, Interval::FALSE]))
                }
                // If both children are uncertain, or if one is certainly true,
                // we cannot conclusively refine their intervals. In this case,
                // propagation does not result in any interval changes.
                else {
                    Ok(Some(vec![]))
                }
            } else {
                // An uncertain logical disjunction result can not shrink the
                // end-points of its children.
                Ok(Some(vec![]))
            }
        } else if self.op.supports_propagation() {
            Ok(
                propagate_comparison(&self.op, interval, left_interval, right_interval)?
                    .map(|(left, right)| vec![left, right]),
            )
        } else {
            Ok(
                propagate_arithmetic(&self.op, interval, left_interval, right_interval)?
                    .map(|(left, right)| vec![left, right]),
            )
        }
    }

    #[expect(deprecated)]
    fn evaluate_statistics(&self, children: &[&Distribution]) -> Result<Distribution> {
        let (left, right) = (children[0], children[1]);

        if self.op.is_numerical_operators() {
            // We might be able to construct the output statistics more accurately,
            // without falling back to an unknown distribution, if we are dealing
            // with Gaussian distributions and numerical operations.
            if let (Gaussian(left), Gaussian(right)) = (left, right)
                && let Some(result) = combine_gaussians(&self.op, left, right)?
            {
                return Ok(Gaussian(result));
            }
        } else if self.op.is_logic_operator() {
            // If we are dealing with logical operators, we expect (and can only
            // operate on) Bernoulli distributions.
            return if let (Bernoulli(left), Bernoulli(right)) = (left, right) {
                combine_bernoullis(&self.op, left, right).map(Bernoulli)
            } else {
                internal_err!(
                    "Logical operators are only compatible with `Bernoulli` distributions"
                )
            };
        } else if self.op.supports_propagation() {
            // If we are handling comparison operators, we expect (and can only
            // operate on) numeric distributions.
            return create_bernoulli_from_comparison(&self.op, left, right);
        }
        // Fall back to an unknown distribution with only summary statistics:
        new_generic_from_binary_op(&self.op, left, right)
    }

    /// For each operator, [`BinaryExpr`] has distinct rules.
    /// TODO: There may be rules specific to some data types and expression ranges.
    fn get_properties(&self, children: &[ExprProperties]) -> Result<ExprProperties> {
        let (l_order, l_range) = (children[0].sort_properties, &children[0].range);
        let (r_order, r_range) = (children[1].sort_properties, &children[1].range);
        match self.op() {
            Operator::Plus => {
                let range = l_range.add(r_range)?;
                Ok(ExprProperties {
                    sort_properties: self.arithmetic_sort_properties(
                        l_order.add(&r_order),
                        l_range,
                        r_range,
                        &range,
                    ),
                    range,
                    preserves_lex_ordering: false,
                    strictly_order_preserving: false,
                })
            }
            Operator::Minus => {
                let range = l_range.sub(r_range)?;
                Ok(ExprProperties {
                    sort_properties: self.arithmetic_sort_properties(
                        l_order.sub(&r_order),
                        l_range,
                        r_range,
                        &range,
                    ),
                    range,
                    preserves_lex_ordering: false,
                    strictly_order_preserving: false,
                })
            }
            Operator::Gt => Ok(ExprProperties {
                sort_properties: l_order.gt_or_gteq(&r_order),
                range: l_range.gt(r_range)?,
                preserves_lex_ordering: false,
                strictly_order_preserving: false,
            }),
            Operator::GtEq => Ok(ExprProperties {
                sort_properties: l_order.gt_or_gteq(&r_order),
                range: l_range.gt_eq(r_range)?,
                preserves_lex_ordering: false,
                strictly_order_preserving: false,
            }),
            Operator::Lt => Ok(ExprProperties {
                sort_properties: r_order.gt_or_gteq(&l_order),
                range: l_range.lt(r_range)?,
                preserves_lex_ordering: false,
                strictly_order_preserving: false,
            }),
            Operator::LtEq => Ok(ExprProperties {
                sort_properties: r_order.gt_or_gteq(&l_order),
                range: l_range.lt_eq(r_range)?,
                preserves_lex_ordering: false,
                strictly_order_preserving: false,
            }),
            Operator::And => Ok(ExprProperties {
                sort_properties: r_order.and_or(&l_order),
                range: l_range.and(r_range)?,
                preserves_lex_ordering: false,
                strictly_order_preserving: false,
            }),
            Operator::Or => Ok(ExprProperties {
                sort_properties: r_order.and_or(&l_order),
                range: l_range.or(r_range)?,
                preserves_lex_ordering: false,
                strictly_order_preserving: false,
            }),
            _ => Ok(ExprProperties::new_unknown()),
        }
    }

    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        fn write_child(
            f: &mut std::fmt::Formatter,
            expr: &dyn PhysicalExpr,
            precedence: u8,
        ) -> std::fmt::Result {
            if let Some(child) = expr.downcast_ref::<BinaryExpr>() {
                let p = child.op.precedence();
                if p == 0 || p < precedence {
                    write!(f, "(")?;
                    child.fmt_sql(f)?;
                    write!(f, ")")
                } else {
                    child.fmt_sql(f)
                }
            } else {
                expr.fmt_sql(f)
            }
        }

        let precedence = self.op.precedence();
        write_child(f, self.left.as_ref(), precedence)?;
        write!(f, " {} ", self.op)?;
        write_child(f, self.right.as_ref(), precedence)
    }

    #[cfg(feature = "proto")]
    fn try_to_proto(
        &self,
        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
        use datafusion_proto_models::protobuf;

        // Linearize a nested binary expression tree of the same operator
        // into a flat vector of operands to avoid deep recursion in proto.
        let op = self.op;
        let mut operand_refs: Vec<&Arc<dyn PhysicalExpr>> = vec![&self.right];
        let mut current_expr: &BinaryExpr = self;
        loop {
            match current_expr.left.downcast_ref::<BinaryExpr>() {
                Some(bin) if bin.op == op => {
                    operand_refs.push(&bin.right);
                    current_expr = bin;
                }
                _ => {
                    operand_refs.push(&current_expr.left);
                    break;
                }
            }
        }
        // Reverse so operands are ordered from left innermost to right outermost.
        operand_refs.reverse();

        let operands = ctx.encode_children_expressions(operand_refs)?;

        Ok(Some(protobuf::PhysicalExprNode {
            expr_id: None,
            expr_type: Some(protobuf::physical_expr_node::ExprType::BinaryExpr(
                Box::new(protobuf::PhysicalBinaryExprNode {
                    l: None,
                    r: None,
                    op: format!("{op:?}"),
                    operands,
                }),
            )),
        }))
    }
}

#[cfg(feature = "proto")]
impl BinaryExpr {
    /// Reconstruct a [`BinaryExpr`] (or a left-deep tree of them when the proto
    /// uses the linearized `operands` form) from its protobuf representation.
    ///
    /// Takes the whole [`PhysicalExprNode`] — the exact inverse of what
    /// [`PhysicalExpr::try_to_proto`] produces — so every expression's
    /// `try_from_proto` shares one signature. The operator string is parsed
    /// via the canonical [`Operator::from_proto_name`] mapping, so no `op`
    /// argument needs to be threaded in by the caller.
    ///
    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
    /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto
    /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode
    pub fn try_from_proto(
        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
    ) -> Result<Arc<dyn PhysicalExpr>> {
        use datafusion_physical_expr_common::expect_expr_variant;
        use datafusion_proto_models::protobuf;
        let node = expect_expr_variant!(
            node,
            protobuf::physical_expr_node::ExprType::BinaryExpr,
            "BinaryExpr",
        );
        let op = Operator::from_proto_name(&node.op).ok_or_else(|| {
            datafusion_common::DataFusionError::Internal(format!(
                "Unsupported binary operator '{}'",
                node.op
            ))
        })?;

        if !node.operands.is_empty() {
            // New linearized format: reduce the flat operands list back into
            // a nested binary expression tree.
            let operands = ctx.decode_children_expressions(&node.operands)?;

            if operands.len() < 2 {
                return internal_err!(
                    "A binary expression must always have at least 2 operands"
                );
            }

            Ok(operands
                .into_iter()
                .reduce(|left, right| {
                    Arc::new(BinaryExpr::new(left, op, right)) as Arc<dyn PhysicalExpr>
                })
                .expect("Binary expression could not be reduced to a single expression."))
        } else {
            // Legacy format with l/r fields.
            let left =
                ctx.decode_required_expression(node.l.as_deref(), "BinaryExpr", "left")?;
            let right =
                ctx.decode_required_expression(node.r.as_deref(), "BinaryExpr", "right")?;
            Ok(Arc::new(BinaryExpr::new(left, op, right)))
        }
    }
}

/// Casts dictionary array to result type for binary numerical operators. Such operators
/// between array and scalar produce a dictionary array other than primitive array of the
/// same operators between array and array. This leads to inconsistent result types causing
/// errors in the following query execution. For such operators between array and scalar,
/// we cast the dictionary array to primitive array.
fn to_result_type_array(
    op: &Operator,
    array: ArrayRef,
    result_type: &DataType,
) -> Result<ArrayRef> {
    if array.data_type() == result_type {
        Ok(array)
    } else if op.is_numerical_operators() {
        match array.data_type() {
            DataType::Dictionary(_, value_type) => {
                if value_type.as_ref() == result_type {
                    Ok(cast(&array, result_type)?)
                } else {
                    internal_err!(
                        "Incompatible Dictionary value type {value_type} with result type {result_type} of Binary operator {op:?}"
                    )
                }
            }
            _ => Ok(array),
        }
    } else {
        Ok(array)
    }
}

impl BinaryExpr {
    /// Evaluate the expression of the left input is an array and
    /// right is literal - use scalar operations
    fn evaluate_array_scalar(
        &self,
        array: &dyn Array,
        scalar: ScalarValue,
    ) -> Result<Option<Result<ArrayRef>>> {
        use Operator::*;
        let scalar_result = match &self.op {
            RegexMatch => regex_match_dyn_scalar(array, &scalar, false, false),
            RegexIMatch => regex_match_dyn_scalar(array, &scalar, false, true),
            RegexNotMatch => regex_match_dyn_scalar(array, &scalar, true, false),
            RegexNotIMatch => regex_match_dyn_scalar(array, &scalar, true, true),
            BitwiseAnd => bitwise_and_dyn_scalar(array, scalar),
            BitwiseOr => bitwise_or_dyn_scalar(array, scalar),
            BitwiseXor => bitwise_xor_dyn_scalar(array, scalar),
            BitwiseShiftRight => bitwise_shift_right_dyn_scalar(array, scalar),
            BitwiseShiftLeft => bitwise_shift_left_dyn_scalar(array, scalar),
            // if scalar operation is not supported - fallback to array implementation
            _ => None,
        };

        Ok(scalar_result)
    }

    fn evaluate_with_resolved_args(
        &self,
        left: Arc<dyn Array>,
        left_data_type: &DataType,
        right: Arc<dyn Array>,
        right_data_type: &DataType,
    ) -> Result<ArrayRef> {
        use Operator::*;
        match &self.op {
            IsDistinctFrom | IsNotDistinctFrom | Lt | LtEq | Gt | GtEq | Eq | NotEq
            | Plus | Minus | Multiply | Divide | Modulo | LikeMatch | ILikeMatch
            | NotLikeMatch | NotILikeMatch => unreachable!(),
            And => {
                if left_data_type == &DataType::Boolean {
                    Ok(boolean_op(&left, &right, and_kleene)?)
                } else {
                    internal_err!(
                        "Cannot evaluate binary expression {:?} with types {:?} and {:?}",
                        self.op,
                        left.data_type(),
                        right.data_type()
                    )
                }
            }
            Or => {
                if left_data_type == &DataType::Boolean {
                    Ok(boolean_op(&left, &right, or_kleene)?)
                } else {
                    internal_err!(
                        "Cannot evaluate binary expression {:?} with types {:?} and {:?}",
                        self.op,
                        left_data_type,
                        right_data_type
                    )
                }
            }
            RegexMatch => regex_match_dyn(&left, &right, false, false),
            RegexIMatch => regex_match_dyn(&left, &right, false, true),
            RegexNotMatch => regex_match_dyn(&left, &right, true, false),
            RegexNotIMatch => regex_match_dyn(&left, &right, true, true),
            BitwiseAnd => bitwise_and_dyn(left, right),
            BitwiseOr => bitwise_or_dyn(left, right),
            BitwiseXor => bitwise_xor_dyn(left, right),
            BitwiseShiftRight => bitwise_shift_right_dyn(left, right),
            BitwiseShiftLeft => bitwise_shift_left_dyn(left, right),
            StringConcat => concat_elements_dyn(&left, &right).map_err(|e| e.into()),
            AtArrow | ArrowAt | Arrow | LongArrow | HashArrow | HashLongArrow | AtAt
            | HashMinus | AtQuestion | Question | QuestionAnd | QuestionPipe
            | IntegerDivide | Colon => {
                not_impl_err!(
                    "Binary operator '{:?}' is not supported in the physical expr",
                    self.op
                )
            }
        }
    }
}

enum ShortCircuitStrategy {
    None,
    ReturnLeft,
    ReturnRight,
    /// Evaluate the right-hand side only on the rows selected by `mask`, then
    /// scatter the results back, filling the unselected rows with `fill_value`.
    ///
    /// - For `AND`, `mask` selects the rows where the LHS is `true` and
    ///   `fill_value` is `false` (rows where the LHS is `false` are `false`).
    /// - For `OR`, `mask` selects the rows where the LHS is `false` and
    ///   `fill_value` is `true` (rows where the LHS is `true` are `true`).
    PreSelection {
        mask: BooleanArray,
        fill_value: bool,
    },
}

/// Based on the results calculated from the left side of the short-circuit operation,
/// pre-selection filters the `RecordBatch` before evaluating the right-hand side when
/// the side that cannot short-circuit the operator is rare:
/// - for `AND`, when the proportion of `true` is less than or equal to 0.2
/// - for `OR`, when the proportion of `false` is less than or equal to 0.2
const PRE_SELECTION_THRESHOLD: f32 = 0.2;

/// Checks if a logical operator (`AND`/`OR`) can short-circuit evaluation based on the left-hand side (lhs) result.
///
/// Short-circuiting occurs under these circumstances:
/// - For `AND`:
///    - if LHS is all false => short-circuit → return LHS
///    - if LHS is all true  => short-circuit → return RHS
///    - if LHS is mixed and true_count / len <= [`PRE_SELECTION_THRESHOLD`] -> pre-selection
/// - For `OR`:
///    - if LHS is all true  => short-circuit → return LHS
///    - if LHS is all false => short-circuit → return RHS
///    - if LHS is mixed and false_count / len <= [`PRE_SELECTION_THRESHOLD`] -> pre-selection
/// # Arguments
/// * `lhs` - The left-hand side (lhs) columnar value (array or scalar)
/// * `op` - The logical operator (`AND` or `OR`)
///
/// # Implementation Notes
/// 1. Only works with Boolean-typed arguments (other types automatically return `false`)
/// 2. Handles both scalar values and array values
/// 3. For arrays, uses optimized bit counting techniques for boolean arrays
fn check_short_circuit(lhs: &ColumnarValue, op: &Operator) -> ShortCircuitStrategy {
    // Only logical operators can use this path.
    let is_and = match op {
        Operator::And => true,
        Operator::Or => false,
        _ => return ShortCircuitStrategy::None,
    };

    // Non-boolean types can't be short-circuited
    if lhs.data_type() != DataType::Boolean {
        return ShortCircuitStrategy::None;
    }

    match lhs {
        ColumnarValue::Array(array) => {
            // Fast path for arrays - try to downcast to boolean array
            if let Ok(bool_array) = as_boolean_array(array) {
                // Arrays with nulls can't be short-circuited
                if bool_array.null_count() > 0 {
                    return ShortCircuitStrategy::None;
                }

                let len = bool_array.len();
                if len == 0 {
                    return ShortCircuitStrategy::None;
                }

                let true_count = bool_array.values().count_set_bits();
                if is_and {
                    if true_count == 0 {
                        return ShortCircuitStrategy::ReturnLeft;
                    }

                    if true_count == len {
                        return ShortCircuitStrategy::ReturnRight;
                    }

                    if true_count as f32 / len as f32 <= PRE_SELECTION_THRESHOLD {
                        // Select rows where the LHS is true; rows where the LHS
                        // is false are false regardless of the RHS.
                        return ShortCircuitStrategy::PreSelection {
                            mask: bool_array.clone(),
                            fill_value: false,
                        };
                    }
                } else {
                    if true_count == len {
                        return ShortCircuitStrategy::ReturnLeft;
                    }

                    if true_count == 0 {
                        return ShortCircuitStrategy::ReturnRight;
                    }

                    let false_count = len - true_count;
                    if false_count as f32 / len as f32 <= PRE_SELECTION_THRESHOLD {
                        // Select rows where the LHS is false; rows where the LHS
                        // is true are true regardless of the RHS. The LHS has no
                        // nulls here, so negating its bits is infallible.
                        let mask = BooleanArray::new(!bool_array.values(), None);
                        return ShortCircuitStrategy::PreSelection {
                            mask,
                            fill_value: true,
                        };
                    }
                }
            }
        }
        ColumnarValue::Scalar(scalar) => {
            // Fast path for scalar values
            if let ScalarValue::Boolean(Some(is_true)) = scalar {
                // Return Left for:
                // - AND with false value
                // - OR with true value
                if (is_and && !is_true) || (!is_and && *is_true) {
                    return ShortCircuitStrategy::ReturnLeft;
                } else {
                    return ShortCircuitStrategy::ReturnRight;
                }
            }
        }
    }

    // If we can't short-circuit, indicate that normal evaluation should continue
    ShortCircuitStrategy::None
}

/// Collapses a pre-selected expression whose RHS is uniformly `rhs_value` across
/// every selected row, avoiding a scatter:
/// - when it equals `fill_value`, every row is `fill_value` (a scalar);
/// - otherwise the selected rows already equal the RHS, which matches the LHS
///   there, and the unselected rows are the LHS value too, so the result is `lhs`.
fn uniform_pre_selection_result(
    rhs_value: bool,
    fill_value: bool,
    lhs: ColumnarValue,
) -> ColumnarValue {
    if rhs_value == fill_value {
        ColumnarValue::Scalar(ScalarValue::Boolean(Some(fill_value)))
    } else {
        lhs
    }
}

/// Creates a boolean array by scattering compact RHS results into the positions
/// selected by `mask`.
///
/// This function is used for short-circuit evaluation optimization of logical AND/OR operations:
/// - Only selected rows are evaluated on the RHS
/// - Values are copied from `right_result` where `mask` is true
/// - All other positions are filled with `fill_value` (`false` for AND, `true` for OR)
///
/// # Parameters
/// - `mask` Boolean array with the rows whose result depends on the RHS
/// - `right_result` Result of evaluating right side of expression (only for selected positions)
/// - `fill_value` The value for the unselected positions (`false` for AND, `true` for OR)
///
/// # Returns
/// A combined `ColumnarValue` with the same length as `mask`.
fn pre_selection_scatter(
    mask: &BooleanArray,
    right_result: Option<&BooleanArray>,
    fill_value: bool,
) -> Result<ColumnarValue> {
    let result_len = mask.len();

    let mut result_array_builder = BooleanArray::builder(result_len);

    let mut right_array_pos = 0;
    let mut last_end = 0;
    match right_result {
        Some(right_result) => {
            SlicesIterator::new(mask).for_each(|(start, end)| {
                if start > last_end {
                    result_array_builder.append_n(start - last_end, fill_value);
                }

                // copy values from right array for this slice
                let len = end - start;
                right_result
                    .slice(right_array_pos, len)
                    .iter()
                    .for_each(|v| result_array_builder.append_option(v));

                right_array_pos += len;
                last_end = end;
            });
        }
        None => SlicesIterator::new(mask).for_each(|(start, end)| {
            if start > last_end {
                result_array_builder.append_n(start - last_end, fill_value);
            }

            let len = end - start;
            result_array_builder.append_nulls(len);

            last_end = end;
        }),
    }

    // Fill any remaining positions with `fill_value`
    if last_end < result_len {
        result_array_builder.append_n(result_len - last_end, fill_value);
    }
    let boolean_result = result_array_builder.finish();

    Ok(ColumnarValue::Array(Arc::new(boolean_result)))
}

/// Create a binary expression whose arguments are correctly coerced.
/// This function errors if it is not possible to coerce the arguments
/// to computational types supported by the operator.
pub fn binary(
    lhs: Arc<dyn PhysicalExpr>,
    op: Operator,
    rhs: Arc<dyn PhysicalExpr>,
    _input_schema: &Schema,
) -> Result<Arc<dyn PhysicalExpr>> {
    Ok(Arc::new(BinaryExpr::new(lhs, op, rhs)))
}

/// Create a similar to expression
pub fn similar_to(
    negated: bool,
    case_insensitive: bool,
    expr: Arc<dyn PhysicalExpr>,
    pattern: Arc<dyn PhysicalExpr>,
) -> Result<Arc<dyn PhysicalExpr>> {
    let binary_op = match (negated, case_insensitive) {
        (false, false) => Operator::RegexMatch,
        (false, true) => Operator::RegexIMatch,
        (true, false) => Operator::RegexNotMatch,
        (true, true) => Operator::RegexNotIMatch,
    };
    Ok(Arc::new(BinaryExpr::new(expr, binary_op, pattern)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::expressions::{Column, Literal, col, lit, try_cast};
    use datafusion_expr::lit as expr_lit;

    use datafusion_common::{assert_contains, plan_datafusion_err};
    use datafusion_physical_expr_common::physical_expr::fmt_sql;

    use crate::planner::logical2physical;
    use arrow::array::BooleanArray;
    use arrow::compute::SortOptions;
    use datafusion_expr::col as logical_col;

    #[test]
    fn test_arithmetic_ordering_overflow() -> Result<()> {
        let asc = SortProperties::Ordered(Default::default());
        let ordered = |range: Interval| ExprProperties {
            sort_properties: asc,
            range,
            preserves_lex_ordering: false,
            strictly_order_preserving: false,
        };

        let schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let a_plus_b =
            BinaryExpr::new(col("a", &schema)?, Operator::Plus, col("b", &schema)?);
        let unbounded = [
            ordered(Interval::make_unbounded(&DataType::Int32)?),
            ordered(Interval::make_unbounded(&DataType::Int32)?),
        ];
        let bounded = [
            ordered(Interval::make(Some(0), Some(10))?),
            ordered(Interval::make(Some(0), Some(10))?),
        ];

        // Unknown ranges: the sum may overflow and wrap, so it is unordered.
        assert_eq!(
            a_plus_b.get_properties(&unbounded)?.sort_properties,
            SortProperties::Unordered
        );
        // Bounded ranges that cannot overflow keep the ordering, as does
        // checked arithmetic, which errors instead of wrapping.
        assert_eq!(a_plus_b.get_properties(&bounded)?.sort_properties, asc);
        let checked = a_plus_b.with_fail_on_overflow(true);
        assert_eq!(checked.get_properties(&unbounded)?.sort_properties, asc);

        // `time + interval` wraps around the 24-hour clock even in checked
        // mode, so it never preserves ordering.
        let time = DataType::Time64(TimeUnit::Nanosecond);
        let interval = DataType::Interval(IntervalUnit::MonthDayNano);
        let schema = Schema::new(vec![
            Field::new("t", time.clone(), false),
            Field::new("i", interval.clone(), false),
        ]);
        let time_plus_interval =
            BinaryExpr::new(col("t", &schema)?, Operator::Plus, col("i", &schema)?)
                .with_fail_on_overflow(true);
        let time_props = [
            ordered(Interval::make_unbounded(&time)?),
            ordered(Interval::make_unbounded(&interval)?),
        ];
        assert_eq!(
            time_plus_interval
                .get_properties(&time_props)?
                .sort_properties,
            SortProperties::Unordered
        );

        Ok(())
    }

    /// `a - b` only derives an ordering when `a` and `b` are ordered in
    /// opposite directions, so every case below pairs an ascending left-hand
    /// side with a descending right-hand side.
    #[test]
    fn test_subtraction_ordering_overflow() -> Result<()> {
        let asc = SortProperties::Ordered(SortOptions {
            descending: false,
            nulls_first: true,
        });
        let desc = SortProperties::Ordered(SortOptions {
            descending: true,
            nulls_first: true,
        });
        let props = |sort_properties, range| ExprProperties {
            sort_properties,
            range,
            preserves_lex_ordering: false,
            strictly_order_preserving: false,
        };

        let schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let a_minus_b =
            BinaryExpr::new(col("a", &schema)?, Operator::Minus, col("b", &schema)?);

        // Signed minimum: the difference can underflow past `i32::MIN` and
        // wrap around to large positive values.
        let signed_underflow = [
            props(asc, Interval::make(Some(i32::MIN), Some(0))?),
            props(desc, Interval::make(Some(0), Some(i32::MAX))?),
        ];
        assert_eq!(
            a_minus_b.get_properties(&signed_underflow)?.sort_properties,
            SortProperties::Unordered
        );
        // The very same ranges keep the ordering under checked arithmetic,
        // which errors instead of wrapping.
        let checked = a_minus_b.clone().with_fail_on_overflow(true);
        assert_eq!(
            checked.get_properties(&signed_underflow)?.sort_properties,
            asc
        );
        // Ranges whose difference stays inside `Int32` are safe.
        let signed_safe = [
            props(asc, Interval::make(Some(0), Some(10))?),
            props(desc, Interval::make(Some(0), Some(10))?),
        ];
        assert_eq!(a_minus_b.get_properties(&signed_safe)?.sort_properties, asc);

        let schema = Schema::new(vec![
            Field::new("a", DataType::UInt32, false),
            Field::new("b", DataType::UInt32, false),
        ]);
        let a_minus_b =
            BinaryExpr::new(col("a", &schema)?, Operator::Minus, col("b", &schema)?);

        // Unsigned underflow: the ranges overlap, so `0 - 1` wraps to
        // `u32::MAX` even though both operands are bounded.
        let unsigned_underflow = [
            props(asc, Interval::make(Some(0_u32), Some(10_u32))?),
            props(desc, Interval::make(Some(0_u32), Some(10_u32))?),
        ];
        assert_eq!(
            a_minus_b
                .get_properties(&unsigned_underflow)?
                .sort_properties,
            SortProperties::Unordered
        );
        // A left-hand range that always dominates the right-hand one cannot
        // underflow.
        let unsigned_safe = [
            props(asc, Interval::make(Some(10_u32), Some(20_u32))?),
            props(desc, Interval::make(Some(0_u32), Some(5_u32))?),
        ];
        assert_eq!(
            a_minus_b.get_properties(&unsigned_safe)?.sort_properties,
            asc
        );

        // `time - interval` wraps around the 24-hour clock even in checked
        // mode, so it never preserves ordering.
        let time = DataType::Time64(TimeUnit::Nanosecond);
        let interval = DataType::Interval(IntervalUnit::MonthDayNano);
        let schema = Schema::new(vec![
            Field::new("t", time.clone(), false),
            Field::new("i", interval.clone(), false),
        ]);
        let time_minus_interval =
            BinaryExpr::new(col("t", &schema)?, Operator::Minus, col("i", &schema)?)
                .with_fail_on_overflow(true);
        let time_props = [
            props(asc, Interval::make_unbounded(&time)?),
            props(desc, Interval::make_unbounded(&interval)?),
        ];
        assert_eq!(
            time_minus_interval
                .get_properties(&time_props)?
                .sort_properties,
            SortProperties::Unordered
        );

        Ok(())
    }

    /// Performs a binary operation, applying any type coercion necessary
    fn binary_op(
        left: Arc<dyn PhysicalExpr>,
        op: Operator,
        right: Arc<dyn PhysicalExpr>,
        schema: &Schema,
    ) -> Result<Arc<dyn PhysicalExpr>> {
        let left_type = left.data_type(schema)?;
        let right_type = right.data_type(schema)?;
        let (lhs, rhs) =
            BinaryTypeCoercer::new(&left_type, &op, &right_type).get_input_types()?;

        let left_expr = try_cast(left, schema, lhs)?;
        let right_expr = try_cast(right, schema, rhs)?;
        binary(left_expr, op, right_expr, schema)
    }

    #[test]
    fn binary_comparison() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);
        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);

        // expression: "a < b"
        let lt = binary(
            col("a", &schema)?,
            Operator::Lt,
            col("b", &schema)?,
            &schema,
        )?;
        let batch =
            RecordBatch::try_new(Arc::new(schema), vec![Arc::new(a), Arc::new(b)])?;

        let result = lt
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");
        assert_eq!(result.len(), 5);

        let expected = [false, false, true, true, true];
        let result =
            as_boolean_array(&result).expect("failed to downcast to BooleanArray");
        for (i, &expected_item) in expected.iter().enumerate().take(5) {
            assert_eq!(result.value(i), expected_item);
        }

        Ok(())
    }

    #[test]
    fn binary_nested() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let a = Int32Array::from(vec![2, 4, 6, 8, 10]);
        let b = Int32Array::from(vec![2, 5, 4, 8, 8]);

        // expression: "a < b OR a == b"
        let expr = binary(
            binary(
                col("a", &schema)?,
                Operator::Lt,
                col("b", &schema)?,
                &schema,
            )?,
            Operator::Or,
            binary(
                col("a", &schema)?,
                Operator::Eq,
                col("b", &schema)?,
                &schema,
            )?,
            &schema,
        )?;
        let batch =
            RecordBatch::try_new(Arc::new(schema), vec![Arc::new(a), Arc::new(b)])?;

        assert_eq!("a@0 < b@1 OR a@0 = b@1", format!("{expr}"));

        let result = expr
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");
        assert_eq!(result.len(), 5);

        let expected = [true, true, false, true, false];
        let result =
            as_boolean_array(&result).expect("failed to downcast to BooleanArray");
        for (i, &expected_item) in expected.iter().enumerate().take(5) {
            assert_eq!(result.value(i), expected_item);
        }

        Ok(())
    }

    // runs an end-to-end test of physical type coercion:
    // 1. construct a record batch with two columns of type A and B
    //  (*_ARRAY is the Rust Arrow array type, and *_TYPE is the DataType of the elements)
    // 2. construct a physical expression of A OP B
    // 3. evaluate the expression
    // 4. verify that the resulting expression is of type C
    // 5. verify that the results of evaluation are $VEC
    macro_rules! test_coercion {
        ($A_ARRAY:ident, $A_TYPE:expr, $A_VEC:expr, $B_ARRAY:ident, $B_TYPE:expr, $B_VEC:expr, $OP:expr, $C_ARRAY:ident, $C_TYPE:expr, $VEC:expr,) => {{
            let schema = Schema::new(vec![
                Field::new("a", $A_TYPE, false),
                Field::new("b", $B_TYPE, false),
            ]);
            let a = $A_ARRAY::from($A_VEC);
            let b = $B_ARRAY::from($B_VEC);
            let (lhs, rhs) =
                BinaryTypeCoercer::new(&$A_TYPE, &$OP, &$B_TYPE).get_input_types()?;

            let left = try_cast(col("a", &schema)?, &schema, lhs)?;
            let right = try_cast(col("b", &schema)?, &schema, rhs)?;

            // verify that we can construct the expression
            let expression = binary(left, $OP, right, &schema)?;
            let batch = RecordBatch::try_new(
                Arc::new(schema.clone()),
                vec![Arc::new(a), Arc::new(b)],
            )?;

            // verify that the expression's type is correct
            assert_eq!(expression.data_type(&schema)?, $C_TYPE);

            // compute
            let result = expression
                .evaluate(&batch)?
                .into_array(batch.num_rows())
                .expect("Failed to convert to array");

            // verify that the array's data_type is correct
            assert_eq!(*result.data_type(), $C_TYPE);

            // verify that the data itself is downcastable
            let result = result
                .as_any()
                .downcast_ref::<$C_ARRAY>()
                .expect("failed to downcast");
            // verify that the result itself is correct
            for (i, x) in $VEC.iter().enumerate() {
                let v = result.value(i);
                assert_eq!(
                    v, *x,
                    "Unexpected output at position {i}:\n\nActual:\n{v}\n\nExpected:\n{x}"
                );
            }
        }};
    }

    #[test]
    fn test_type_coercion() -> Result<()> {
        test_coercion!(
            Int32Array,
            DataType::Int32,
            vec![1i32, 2i32],
            UInt32Array,
            DataType::UInt32,
            vec![1u32, 2u32],
            Operator::Plus,
            Int64Array,
            DataType::Int64,
            [2i64, 4i64],
        );
        test_coercion!(
            Int32Array,
            DataType::Int32,
            vec![1i32],
            UInt16Array,
            DataType::UInt16,
            vec![1u16],
            Operator::Plus,
            Int32Array,
            DataType::Int32,
            [2i32],
        );
        test_coercion!(
            Float32Array,
            DataType::Float32,
            vec![1f32],
            UInt16Array,
            DataType::UInt16,
            vec![1u16],
            Operator::Plus,
            Float32Array,
            DataType::Float32,
            [2f32],
        );
        test_coercion!(
            Float32Array,
            DataType::Float32,
            vec![2f32],
            UInt16Array,
            DataType::UInt16,
            vec![1u16],
            Operator::Multiply,
            Float32Array,
            DataType::Float32,
            [2f32],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["1994-12-13", "1995-01-26"],
            Date32Array,
            DataType::Date32,
            vec![9112, 9156],
            Operator::Eq,
            BooleanArray,
            DataType::Boolean,
            [true, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["1994-12-13", "1995-01-26"],
            Date32Array,
            DataType::Date32,
            vec![9113, 9154],
            Operator::Lt,
            BooleanArray,
            DataType::Boolean,
            [true, false],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["1994-12-13T12:34:56", "1995-01-26T01:23:45"],
            Date64Array,
            DataType::Date64,
            vec![787322096000, 791083425000],
            Operator::Eq,
            BooleanArray,
            DataType::Boolean,
            [true, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["1994-12-13T12:34:56", "1995-01-26T01:23:45"],
            Date64Array,
            DataType::Date64,
            vec![787322096001, 791083424999],
            Operator::Lt,
            BooleanArray,
            DataType::Boolean,
            [true, false],
        );
        test_coercion!(
            StringViewArray,
            DataType::Utf8View,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexMatch,
            BooleanArray,
            DataType::Boolean,
            [true, false, true, false, false],
        );
        test_coercion!(
            StringViewArray,
            DataType::Utf8View,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexIMatch,
            BooleanArray,
            DataType::Boolean,
            [true, true, true, true, false],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringViewArray,
            DataType::Utf8View,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexNotMatch,
            BooleanArray,
            DataType::Boolean,
            [false, true, false, true, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringViewArray,
            DataType::Utf8View,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexNotIMatch,
            BooleanArray,
            DataType::Boolean,
            [false, false, false, false, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexMatch,
            BooleanArray,
            DataType::Boolean,
            [true, false, true, false, false],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexIMatch,
            BooleanArray,
            DataType::Boolean,
            [true, true, true, true, false],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexNotMatch,
            BooleanArray,
            DataType::Boolean,
            [false, true, false, true, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexNotIMatch,
            BooleanArray,
            DataType::Boolean,
            [false, false, false, false, true],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexMatch,
            BooleanArray,
            DataType::Boolean,
            [true, false, true, false, false],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexIMatch,
            BooleanArray,
            DataType::Boolean,
            [true, true, true, true, false],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexNotMatch,
            BooleanArray,
            DataType::Boolean,
            [false, true, false, true, true],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["^a", "^A", "(b|d)", "(B|D)", "^(b|c)"],
            Operator::RegexNotIMatch,
            BooleanArray,
            DataType::Boolean,
            [false, false, false, false, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::LikeMatch,
            BooleanArray,
            DataType::Boolean,
            [true, false, false, true, false],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::ILikeMatch,
            BooleanArray,
            DataType::Boolean,
            [true, true, false, true, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::NotLikeMatch,
            BooleanArray,
            DataType::Boolean,
            [false, true, true, false, true],
        );
        test_coercion!(
            StringArray,
            DataType::Utf8,
            vec!["abc"; 5],
            StringArray,
            DataType::Utf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::NotILikeMatch,
            BooleanArray,
            DataType::Boolean,
            [false, false, true, false, false],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::LikeMatch,
            BooleanArray,
            DataType::Boolean,
            [true, false, false, true, false],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::ILikeMatch,
            BooleanArray,
            DataType::Boolean,
            [true, true, false, true, true],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::NotLikeMatch,
            BooleanArray,
            DataType::Boolean,
            [false, true, true, false, true],
        );
        test_coercion!(
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["abc"; 5],
            LargeStringArray,
            DataType::LargeUtf8,
            vec!["a__", "A%BC", "A_BC", "abc", "a%C"],
            Operator::NotILikeMatch,
            BooleanArray,
            DataType::Boolean,
            [false, false, true, false, false],
        );
        test_coercion!(
            Int16Array,
            DataType::Int16,
            vec![1i16, 2i16, 3i16],
            Int64Array,
            DataType::Int64,
            vec![10i64, 4i64, 5i64],
            Operator::BitwiseAnd,
            Int64Array,
            DataType::Int64,
            [0i64, 0i64, 1i64],
        );
        test_coercion!(
            UInt16Array,
            DataType::UInt16,
            vec![1u16, 2u16, 3u16],
            UInt64Array,
            DataType::UInt64,
            vec![10u64, 4u64, 5u64],
            Operator::BitwiseAnd,
            UInt64Array,
            DataType::UInt64,
            [0u64, 0u64, 1u64],
        );
        test_coercion!(
            Int16Array,
            DataType::Int16,
            vec![3i16, 2i16, 3i16],
            Int64Array,
            DataType::Int64,
            vec![10i64, 6i64, 5i64],
            Operator::BitwiseOr,
            Int64Array,
            DataType::Int64,
            [11i64, 6i64, 7i64],
        );
        test_coercion!(
            UInt16Array,
            DataType::UInt16,
            vec![1u16, 2u16, 3u16],
            UInt64Array,
            DataType::UInt64,
            vec![10u64, 4u64, 5u64],
            Operator::BitwiseOr,
            UInt64Array,
            DataType::UInt64,
            [11u64, 6u64, 7u64],
        );
        test_coercion!(
            Int16Array,
            DataType::Int16,
            vec![3i16, 2i16, 3i16],
            Int64Array,
            DataType::Int64,
            vec![10i64, 6i64, 5i64],
            Operator::BitwiseXor,
            Int64Array,
            DataType::Int64,
            [9i64, 4i64, 6i64],
        );
        test_coercion!(
            UInt16Array,
            DataType::UInt16,
            vec![3u16, 2u16, 3u16],
            UInt64Array,
            DataType::UInt64,
            vec![10u64, 6u64, 5u64],
            Operator::BitwiseXor,
            UInt64Array,
            DataType::UInt64,
            [9u64, 4u64, 6u64],
        );
        test_coercion!(
            Int16Array,
            DataType::Int16,
            vec![4i16, 27i16, 35i16],
            Int64Array,
            DataType::Int64,
            vec![2i64, 3i64, 4i64],
            Operator::BitwiseShiftRight,
            Int64Array,
            DataType::Int64,
            [1i64, 3i64, 2i64],
        );
        test_coercion!(
            UInt16Array,
            DataType::UInt16,
            vec![4u16, 27u16, 35u16],
            UInt64Array,
            DataType::UInt64,
            vec![2u64, 3u64, 4u64],
            Operator::BitwiseShiftRight,
            UInt64Array,
            DataType::UInt64,
            [1u64, 3u64, 2u64],
        );
        test_coercion!(
            Int16Array,
            DataType::Int16,
            vec![2i16, 3i16, 4i16],
            Int64Array,
            DataType::Int64,
            vec![4i64, 12i64, 7i64],
            Operator::BitwiseShiftLeft,
            Int64Array,
            DataType::Int64,
            [32i64, 12288i64, 512i64],
        );
        test_coercion!(
            UInt16Array,
            DataType::UInt16,
            vec![2u16, 3u16, 4u16],
            UInt64Array,
            DataType::UInt64,
            vec![4u64, 12u64, 7u64],
            Operator::BitwiseShiftLeft,
            UInt64Array,
            DataType::UInt64,
            [32u64, 12288u64, 512u64],
        );
        Ok(())
    }

    // Note it would be nice to use the same test_coercion macro as
    // above, but sadly the type of the values of the dictionary are
    // not encoded in the rust type of the DictionaryArray. Thus there
    // is no way at the time of this writing to create a dictionary
    // array using the `From` trait
    #[test]
    fn test_dictionary_type_to_array_coercion() -> Result<()> {
        // Test string  a string dictionary
        let dict_type =
            DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
        let string_type = DataType::Utf8;

        // build dictionary
        let mut dict_builder = StringDictionaryBuilder::<Int32Type>::new();

        dict_builder.append("one")?;
        dict_builder.append_null();
        dict_builder.append("three")?;
        dict_builder.append("four")?;
        let dict_array = Arc::new(dict_builder.finish()) as ArrayRef;

        let str_array = Arc::new(StringArray::from(vec![
            Some("not one"),
            Some("two"),
            None,
            Some("four"),
        ])) as ArrayRef;

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", dict_type.clone(), true),
            Field::new("b", string_type.clone(), true),
        ]));

        // Test 1: a = b
        let result = BooleanArray::from(vec![Some(false), None, None, Some(true)]);
        apply_logic_op(&schema, &dict_array, &str_array, Operator::Eq, result)?;

        // Test 2: now test the other direction
        // b = a
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", string_type, true),
            Field::new("b", dict_type, true),
        ]));
        let result = BooleanArray::from(vec![Some(false), None, None, Some(true)]);
        apply_logic_op(&schema, &str_array, &dict_array, Operator::Eq, result)?;

        Ok(())
    }

    #[test]
    fn plus_op() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);
        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);

        apply_arithmetic::<Int32Type>(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Plus,
            Int32Array::from(vec![2, 4, 7, 12, 21]),
        )?;

        Ok(())
    }

    #[test]
    fn plus_op_dict() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
        ]);

        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);
        let keys = Int8Array::from(vec![Some(0), None, Some(1), Some(3), None]);
        let a = DictionaryArray::try_new(keys, Arc::new(a))?;

        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);
        let keys = Int8Array::from(vec![0, 1, 1, 2, 1]);
        let b = DictionaryArray::try_new(keys, Arc::new(b))?;

        apply_arithmetic::<Int32Type>(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Plus,
            Int32Array::from(vec![Some(2), None, Some(4), Some(8), None]),
        )?;

        Ok(())
    }

    #[test]
    fn plus_op_dict_decimal() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
        ]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value),
                Some(value + 2),
                Some(value - 1),
                Some(value + 1),
            ],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![Some(0), Some(2), None, Some(3), Some(0)]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let keys = Int8Array::from(vec![Some(0), None, Some(3), Some(2), Some(2)]);
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value + 1),
                Some(value + 3),
                Some(value),
                Some(value + 2),
            ],
            10,
            0,
        ));
        let b = DictionaryArray::try_new(keys, decimal_array)?;

        apply_arithmetic(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Plus,
            create_decimal_array(&[Some(247), None, None, Some(247), Some(246)], 11, 0),
        )?;

        Ok(())
    }

    #[test]
    fn plus_op_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Plus,
            ScalarValue::Int32(Some(1)),
            Arc::new(Int32Array::from(vec![2, 3, 4, 5, 6])),
        )?;

        Ok(())
    }

    #[test]
    fn plus_op_dict_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
            true,
        )]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;

        let a = dict_builder.finish();

        let expected: PrimitiveArray<Int32Type> =
            PrimitiveArray::from(vec![Some(2), None, Some(3), Some(6)]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Plus,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Int32(Some(1))),
            ),
            Arc::new(expected),
        )?;

        Ok(())
    }

    #[test]
    fn plus_op_dict_scalar_decimal() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(
                Box::new(DataType::Int8),
                Box::new(DataType::Decimal128(10, 0)),
            ),
            true,
        )]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[Some(value), None, Some(value - 1), Some(value + 1)],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![0, 2, 1, 3, 0]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value + 1),
                Some(value),
                None,
                Some(value + 2),
                Some(value + 1),
            ],
            11,
            0,
        ));

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Plus,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Decimal128(Some(1), 10, 0)),
            ),
            decimal_array,
        )?;

        Ok(())
    }

    #[test]
    fn minus_op() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]));
        let a = Arc::new(Int32Array::from(vec![1, 2, 4, 8, 16]));
        let b = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));

        apply_arithmetic::<Int32Type>(
            Arc::clone(&schema),
            vec![
                Arc::clone(&a) as Arc<dyn Array>,
                Arc::clone(&b) as Arc<dyn Array>,
            ],
            Operator::Minus,
            Int32Array::from(vec![0, 0, 1, 4, 11]),
        )?;

        // should handle have negative values in result (for signed)
        apply_arithmetic::<Int32Type>(
            schema,
            vec![b, a],
            Operator::Minus,
            Int32Array::from(vec![0, 0, -1, -4, -11]),
        )?;

        Ok(())
    }

    #[test]
    fn date32_minus_date32_returns_int64_days() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Date32, true),
            Field::new("b", DataType::Date32, true),
        ]));
        let a = Arc::new(Date32Array::from(vec![
            Some(18_901),
            Some(18_901),
            None,
            Some(18_900),
        ]));
        let b = Arc::new(Date32Array::from(vec![
            Some(18_898),
            Some(18_904),
            Some(18_900),
            None,
        ]));

        apply_arithmetic::<Int64Type>(
            schema,
            vec![a, b],
            Operator::Minus,
            Int64Array::from(vec![Some(3), Some(-3), None, None]),
        )?;

        Ok(())
    }

    #[test]
    fn date64_minus_date64_returns_int64_days() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Date64, true),
            Field::new("b", DataType::Date64, true),
        ]));
        let a = Arc::new(Date64Array::from(vec![
            Some(18_901 * MILLIS_PER_DAY),
            Some(18_901 * MILLIS_PER_DAY),
            None,
            Some(18_900 * MILLIS_PER_DAY),
        ]));
        let b = Arc::new(Date64Array::from(vec![
            Some(18_898 * MILLIS_PER_DAY),
            Some(18_904 * MILLIS_PER_DAY),
            Some(18_900 * MILLIS_PER_DAY),
            None,
        ]));

        apply_arithmetic::<Int64Type>(
            schema,
            vec![a, b],
            Operator::Minus,
            Int64Array::from(vec![Some(3), Some(-3), None, None]),
        )?;

        Ok(())
    }

    #[test]
    fn date32_minus_null_scalar_returns_int64_null_scalar() -> Result<()> {
        let result = apply_date_subtraction(
            &ColumnarValue::Array(Arc::new(Date32Array::from(vec![
                Some(18_901),
                Some(18_900),
            ]))),
            &ColumnarValue::Scalar(ScalarValue::Date32(None)),
        )?;

        assert!(matches!(
            result,
            ColumnarValue::Scalar(ScalarValue::Int64(None))
        ));

        Ok(())
    }

    #[test]
    fn minus_op_dict() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
        ]);

        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);
        let keys = Int8Array::from(vec![Some(0), None, Some(1), Some(3), None]);
        let a = DictionaryArray::try_new(keys, Arc::new(a))?;

        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);
        let keys = Int8Array::from(vec![0, 1, 1, 2, 1]);
        let b = DictionaryArray::try_new(keys, Arc::new(b))?;

        apply_arithmetic::<Int32Type>(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Minus,
            Int32Array::from(vec![Some(0), None, Some(0), Some(0), None]),
        )?;

        Ok(())
    }

    #[test]
    fn minus_op_dict_decimal() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
        ]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value),
                Some(value + 2),
                Some(value - 1),
                Some(value + 1),
            ],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![Some(0), Some(2), None, Some(3), Some(0)]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let keys = Int8Array::from(vec![Some(0), None, Some(3), Some(2), Some(2)]);
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value + 1),
                Some(value + 3),
                Some(value),
                Some(value + 2),
            ],
            10,
            0,
        ));
        let b = DictionaryArray::try_new(keys, decimal_array)?;

        apply_arithmetic(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Minus,
            create_decimal_array(&[Some(-1), None, None, Some(1), Some(0)], 11, 0),
        )?;

        Ok(())
    }

    #[test]
    fn minus_op_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Minus,
            ScalarValue::Int32(Some(1)),
            Arc::new(Int32Array::from(vec![0, 1, 2, 3, 4])),
        )?;

        Ok(())
    }

    #[test]
    fn minus_op_dict_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
            true,
        )]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;

        let a = dict_builder.finish();

        let expected: PrimitiveArray<Int32Type> =
            PrimitiveArray::from(vec![Some(0), None, Some(1), Some(4)]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Minus,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Int32(Some(1))),
            ),
            Arc::new(expected),
        )?;

        Ok(())
    }

    #[test]
    fn minus_op_dict_scalar_decimal() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(
                Box::new(DataType::Int8),
                Box::new(DataType::Decimal128(10, 0)),
            ),
            true,
        )]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[Some(value), None, Some(value - 1), Some(value + 1)],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![0, 2, 1, 3, 0]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value - 1),
                Some(value - 2),
                None,
                Some(value),
                Some(value - 1),
            ],
            11,
            0,
        ));

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Minus,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Decimal128(Some(1), 10, 0)),
            ),
            decimal_array,
        )?;

        Ok(())
    }

    #[test]
    fn multiply_op() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]));
        let a = Arc::new(Int32Array::from(vec![4, 8, 16, 32, 64]));
        let b = Arc::new(Int32Array::from(vec![2, 4, 8, 16, 32]));

        apply_arithmetic::<Int32Type>(
            schema,
            vec![a, b],
            Operator::Multiply,
            Int32Array::from(vec![8, 32, 128, 512, 2048]),
        )?;

        Ok(())
    }

    #[test]
    fn multiply_op_dict() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
        ]);

        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);
        let keys = Int8Array::from(vec![Some(0), None, Some(1), Some(3), None]);
        let a = DictionaryArray::try_new(keys, Arc::new(a))?;

        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);
        let keys = Int8Array::from(vec![0, 1, 1, 2, 1]);
        let b = DictionaryArray::try_new(keys, Arc::new(b))?;

        apply_arithmetic::<Int32Type>(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Multiply,
            Int32Array::from(vec![Some(1), None, Some(4), Some(16), None]),
        )?;

        Ok(())
    }

    #[test]
    fn multiply_op_dict_decimal() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
        ]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value),
                Some(value + 2),
                Some(value - 1),
                Some(value + 1),
            ],
            10,
            0,
        )) as ArrayRef;

        let keys = Int8Array::from(vec![Some(0), Some(2), None, Some(3), Some(0)]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let keys = Int8Array::from(vec![Some(0), None, Some(3), Some(2), Some(2)]);
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value + 1),
                Some(value + 3),
                Some(value),
                Some(value + 2),
            ],
            10,
            0,
        ));
        let b = DictionaryArray::try_new(keys, decimal_array)?;

        apply_arithmetic(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Multiply,
            create_decimal_array(
                &[Some(15252), None, None, Some(15252), Some(15129)],
                21,
                0,
            ),
        )?;

        Ok(())
    }

    #[test]
    fn multiply_op_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Multiply,
            ScalarValue::Int32(Some(2)),
            Arc::new(Int32Array::from(vec![2, 4, 6, 8, 10])),
        )?;

        Ok(())
    }

    #[test]
    fn multiply_op_dict_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
            true,
        )]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;

        let a = dict_builder.finish();

        let expected: PrimitiveArray<Int32Type> =
            PrimitiveArray::from(vec![Some(2), None, Some(4), Some(10)]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Multiply,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Int32(Some(2))),
            ),
            Arc::new(expected),
        )?;

        Ok(())
    }

    #[test]
    fn multiply_op_dict_scalar_decimal() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(
                Box::new(DataType::Int8),
                Box::new(DataType::Decimal128(10, 0)),
            ),
            true,
        )]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[Some(value), None, Some(value - 1), Some(value + 1)],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![0, 2, 1, 3, 0]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let decimal_array = Arc::new(create_decimal_array(
            &[Some(246), Some(244), None, Some(248), Some(246)],
            21,
            0,
        ));

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Multiply,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Decimal128(Some(2), 10, 0)),
            ),
            decimal_array,
        )?;

        Ok(())
    }

    #[test]
    fn divide_op() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]));
        let a = Arc::new(Int32Array::from(vec![8, 32, 128, 512, 2048]));
        let b = Arc::new(Int32Array::from(vec![2, 4, 8, 16, 32]));

        apply_arithmetic::<Int32Type>(
            schema,
            vec![a, b],
            Operator::Divide,
            Int32Array::from(vec![4, 8, 16, 32, 64]),
        )?;

        Ok(())
    }

    #[test]
    fn divide_op_dict() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
        ]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;
        dict_builder.append(0)?;

        let a = dict_builder.finish();

        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);
        let keys = Int8Array::from(vec![0, 1, 1, 2, 1]);
        let b = DictionaryArray::try_new(keys, Arc::new(b))?;

        apply_arithmetic::<Int32Type>(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Divide,
            Int32Array::from(vec![Some(1), None, Some(1), Some(1), Some(0)]),
        )?;

        Ok(())
    }

    #[test]
    fn divide_op_dict_decimal() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
        ]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value),
                Some(value + 2),
                Some(value - 1),
                Some(value + 1),
            ],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![Some(0), Some(2), None, Some(3), Some(0)]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let keys = Int8Array::from(vec![Some(0), None, Some(3), Some(2), Some(2)]);
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value + 1),
                Some(value + 3),
                Some(value),
                Some(value + 2),
            ],
            10,
            0,
        ));
        let b = DictionaryArray::try_new(keys, decimal_array)?;

        apply_arithmetic(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Divide,
            create_decimal_array(
                &[
                    Some(9919), // 0.9919
                    None,
                    None,
                    Some(10081), // 1.0081
                    Some(10000), // 1.0
                ],
                14,
                4,
            ),
        )?;

        Ok(())
    }

    #[test]
    fn divide_op_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Divide,
            ScalarValue::Int32(Some(2)),
            Arc::new(Int32Array::from(vec![0, 1, 1, 2, 2])),
        )?;

        Ok(())
    }

    #[test]
    fn divide_op_dict_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
            true,
        )]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;

        let a = dict_builder.finish();

        let expected: PrimitiveArray<Int32Type> =
            PrimitiveArray::from(vec![Some(0), None, Some(1), Some(2)]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Divide,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Int32(Some(2))),
            ),
            Arc::new(expected),
        )?;

        Ok(())
    }

    #[test]
    fn divide_op_dict_scalar_decimal() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(
                Box::new(DataType::Int8),
                Box::new(DataType::Decimal128(10, 0)),
            ),
            true,
        )]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[Some(value), None, Some(value - 1), Some(value + 1)],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![0, 2, 1, 3, 0]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let decimal_array = Arc::new(create_decimal_array(
            &[Some(615000), Some(610000), None, Some(620000), Some(615000)],
            14,
            4,
        ));

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Divide,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Decimal128(Some(2), 10, 0)),
            ),
            decimal_array,
        )?;

        Ok(())
    }

    #[test]
    fn modulus_op() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]));
        let a = Arc::new(Int32Array::from(vec![8, 32, 128, 512, 2048]));
        let b = Arc::new(Int32Array::from(vec![2, 4, 7, 14, 32]));

        apply_arithmetic::<Int32Type>(
            schema,
            vec![a, b],
            Operator::Modulo,
            Int32Array::from(vec![0, 0, 2, 8, 0]),
        )?;

        Ok(())
    }

    #[test]
    fn modulus_op_dict() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
                true,
            ),
        ]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;
        dict_builder.append(0)?;

        let a = dict_builder.finish();

        let b = Int32Array::from(vec![1, 2, 4, 8, 16]);
        let keys = Int8Array::from(vec![0, 1, 1, 2, 1]);
        let b = DictionaryArray::try_new(keys, Arc::new(b))?;

        apply_arithmetic::<Int32Type>(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Modulo,
            Int32Array::from(vec![Some(0), None, Some(0), Some(1), Some(0)]),
        )?;

        Ok(())
    }

    #[test]
    fn modulus_op_dict_decimal() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new(
                "a",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
            Field::new(
                "b",
                DataType::Dictionary(
                    Box::new(DataType::Int8),
                    Box::new(DataType::Decimal128(10, 0)),
                ),
                true,
            ),
        ]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value),
                Some(value + 2),
                Some(value - 1),
                Some(value + 1),
            ],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![Some(0), Some(2), None, Some(3), Some(0)]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let keys = Int8Array::from(vec![Some(0), None, Some(3), Some(2), Some(2)]);
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value + 1),
                Some(value + 3),
                Some(value),
                Some(value + 2),
            ],
            10,
            0,
        ));
        let b = DictionaryArray::try_new(keys, decimal_array)?;

        apply_arithmetic(
            Arc::new(schema),
            vec![Arc::new(a), Arc::new(b)],
            Operator::Modulo,
            create_decimal_array(&[Some(123), None, None, Some(1), Some(0)], 10, 0),
        )?;

        Ok(())
    }

    #[test]
    fn modulus_op_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
        let a = Int32Array::from(vec![1, 2, 3, 4, 5]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Modulo,
            ScalarValue::Int32(Some(2)),
            Arc::new(Int32Array::from(vec![1, 0, 1, 0, 1])),
        )?;

        Ok(())
    }

    #[test]
    fn modules_op_dict_scalar() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Int32)),
            true,
        )]);

        let mut dict_builder = PrimitiveDictionaryBuilder::<Int8Type, Int32Type>::new();

        dict_builder.append(1)?;
        dict_builder.append_null();
        dict_builder.append(2)?;
        dict_builder.append(5)?;

        let a = dict_builder.finish();

        let expected: PrimitiveArray<Int32Type> =
            PrimitiveArray::from(vec![Some(1), None, Some(0), Some(1)]);

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Modulo,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Int32(Some(2))),
            ),
            Arc::new(expected),
        )?;

        Ok(())
    }

    #[test]
    fn modulus_op_dict_scalar_decimal() -> Result<()> {
        let schema = Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(
                Box::new(DataType::Int8),
                Box::new(DataType::Decimal128(10, 0)),
            ),
            true,
        )]);

        let value = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[Some(value), None, Some(value - 1), Some(value + 1)],
            10,
            0,
        ));

        let keys = Int8Array::from(vec![0, 2, 1, 3, 0]);
        let a = DictionaryArray::try_new(keys, decimal_array)?;

        let decimal_array = Arc::new(create_decimal_array(
            &[Some(1), Some(0), None, Some(0), Some(1)],
            10,
            0,
        ));

        apply_arithmetic_scalar(
            Arc::new(schema),
            vec![Arc::new(a)],
            Operator::Modulo,
            ScalarValue::Dictionary(
                Box::new(DataType::Int8),
                Box::new(ScalarValue::Decimal128(Some(2), 10, 0)),
            ),
            decimal_array,
        )?;

        Ok(())
    }

    fn apply_arithmetic<T: ArrowNumericType>(
        schema: SchemaRef,
        data: Vec<ArrayRef>,
        op: Operator,
        expected: PrimitiveArray<T>,
    ) -> Result<()> {
        let arithmetic_op =
            binary_op(col("a", &schema)?, op, col("b", &schema)?, &schema)?;
        let batch = RecordBatch::try_new(schema, data)?;
        let result = arithmetic_op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");

        assert_eq!(result.as_ref(), &expected);
        Ok(())
    }

    fn apply_arithmetic_scalar(
        schema: SchemaRef,
        data: Vec<ArrayRef>,
        op: Operator,
        literal: ScalarValue,
        expected: ArrayRef,
    ) -> Result<()> {
        let lit = Arc::new(Literal::new(literal));
        let arithmetic_op = binary_op(col("a", &schema)?, op, lit, &schema)?;
        let batch = RecordBatch::try_new(schema, data)?;
        let result = arithmetic_op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");

        assert_eq!(&result, &expected);
        Ok(())
    }

    fn apply_logic_op(
        schema: &SchemaRef,
        left: &ArrayRef,
        right: &ArrayRef,
        op: Operator,
        expected: BooleanArray,
    ) -> Result<()> {
        let op = binary_op(col("a", schema)?, op, col("b", schema)?, schema)?;
        let data: Vec<ArrayRef> = vec![Arc::clone(left), Arc::clone(right)];
        let batch = RecordBatch::try_new(Arc::clone(schema), data)?;
        let result = op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");

        assert_eq!(result.as_ref(), &expected);
        Ok(())
    }

    // Test `scalar <op> arr` produces expected
    fn apply_logic_op_scalar_arr(
        schema: &SchemaRef,
        scalar: &ScalarValue,
        arr: &ArrayRef,
        op: Operator,
        expected: &BooleanArray,
    ) -> Result<()> {
        let scalar = lit(scalar.clone());
        let op = binary_op(scalar, op, col("a", schema)?, schema)?;
        let batch = RecordBatch::try_new(Arc::clone(schema), vec![Arc::clone(arr)])?;
        let result = op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");
        assert_eq!(result.as_ref(), expected);

        Ok(())
    }

    // Test `arr <op> scalar` produces expected
    fn apply_logic_op_arr_scalar(
        schema: &SchemaRef,
        arr: &ArrayRef,
        scalar: &ScalarValue,
        op: Operator,
        expected: &BooleanArray,
    ) -> Result<()> {
        let scalar = lit(scalar.clone());
        let op = binary_op(col("a", schema)?, op, scalar, schema)?;
        let batch = RecordBatch::try_new(Arc::clone(schema), vec![Arc::clone(arr)])?;
        let result = op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");
        assert_eq!(result.as_ref(), expected);

        Ok(())
    }

    #[test]
    fn and_with_nulls_op() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Boolean, true),
            Field::new("b", DataType::Boolean, true),
        ]);
        let a = Arc::new(BooleanArray::from(vec![
            Some(true),
            Some(false),
            None,
            Some(true),
            Some(false),
            None,
            Some(true),
            Some(false),
            None,
        ])) as ArrayRef;
        let b = Arc::new(BooleanArray::from(vec![
            Some(true),
            Some(true),
            Some(true),
            Some(false),
            Some(false),
            Some(false),
            None,
            None,
            None,
        ])) as ArrayRef;

        let expected = BooleanArray::from(vec![
            Some(true),
            Some(false),
            None,
            Some(false),
            Some(false),
            Some(false),
            None,
            Some(false),
            None,
        ]);
        apply_logic_op(&Arc::new(schema), &a, &b, Operator::And, expected)?;

        Ok(())
    }

    #[test]
    fn regex_with_nulls() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Utf8, true),
            Field::new("b", DataType::Utf8, true),
        ]);
        let a = Arc::new(StringArray::from(vec![
            Some("abc"),
            None,
            Some("abc"),
            None,
            Some("abc"),
        ])) as ArrayRef;
        let b = Arc::new(StringArray::from(vec![
            Some("^a"),
            Some("^A"),
            None,
            None,
            Some("^(b|c)"),
        ])) as ArrayRef;

        let regex_expected =
            BooleanArray::from(vec![Some(true), None, None, None, Some(false)]);
        let regex_not_expected =
            BooleanArray::from(vec![Some(false), None, None, None, Some(true)]);
        apply_logic_op(
            &Arc::new(schema.clone()),
            &a,
            &b,
            Operator::RegexMatch,
            regex_expected.clone(),
        )?;
        apply_logic_op(
            &Arc::new(schema.clone()),
            &a,
            &b,
            Operator::RegexIMatch,
            regex_expected.clone(),
        )?;
        apply_logic_op(
            &Arc::new(schema.clone()),
            &a,
            &b,
            Operator::RegexNotMatch,
            regex_not_expected.clone(),
        )?;
        apply_logic_op(
            &Arc::new(schema),
            &a,
            &b,
            Operator::RegexNotIMatch,
            regex_not_expected.clone(),
        )?;

        let schema = Schema::new(vec![
            Field::new("a", DataType::LargeUtf8, true),
            Field::new("b", DataType::LargeUtf8, true),
        ]);
        let a = Arc::new(LargeStringArray::from(vec![
            Some("abc"),
            None,
            Some("abc"),
            None,
            Some("abc"),
        ])) as ArrayRef;
        let b = Arc::new(LargeStringArray::from(vec![
            Some("^a"),
            Some("^A"),
            None,
            None,
            Some("^(b|c)"),
        ])) as ArrayRef;

        apply_logic_op(
            &Arc::new(schema.clone()),
            &a,
            &b,
            Operator::RegexMatch,
            regex_expected.clone(),
        )?;
        apply_logic_op(
            &Arc::new(schema.clone()),
            &a,
            &b,
            Operator::RegexIMatch,
            regex_expected,
        )?;
        apply_logic_op(
            &Arc::new(schema.clone()),
            &a,
            &b,
            Operator::RegexNotMatch,
            regex_not_expected.clone(),
        )?;
        apply_logic_op(
            &Arc::new(schema),
            &a,
            &b,
            Operator::RegexNotIMatch,
            regex_not_expected,
        )?;

        Ok(())
    }

    #[test]
    fn regex_scalar_with_dictionary_nulls() -> Result<()> {
        let dictionary_values = Arc::new(StringArray::from(vec![
            Some("abc"),
            None,
            Some("ABC"),
            Some("def"),
        ]));
        let keys = UInt32Array::from(vec![Some(0), None, Some(1), Some(2), Some(3)]);
        let dictionary =
            Arc::new(DictionaryArray::try_new(keys, dictionary_values)?) as ArrayRef;
        let utf8 = cast(&dictionary, &DataType::Utf8)?;
        let pattern = ScalarValue::Utf8(Some("^abc$".to_string()));
        let dictionary_schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            dictionary.data_type().clone(),
            true,
        )]));
        let utf8_schema =
            Arc::new(Schema::new(vec![Field::new("a", DataType::Utf8, true)]));

        let evaluate =
            |schema: &SchemaRef, array: &ArrayRef, op: Operator| -> Result<ArrayRef> {
                let expr = binary(col("a", schema)?, op, lit(pattern.clone()), schema)?;
                let batch =
                    RecordBatch::try_new(Arc::clone(schema), vec![Arc::clone(array)])?;
                Ok(expr
                    .evaluate(&batch)?
                    .into_array(batch.num_rows())
                    .expect("Failed to convert to array"))
            };

        for (op, expected) in [
            (
                Operator::RegexMatch,
                BooleanArray::from(vec![
                    Some(true),
                    None,
                    None,
                    Some(false),
                    Some(false),
                ]),
            ),
            (
                Operator::RegexIMatch,
                BooleanArray::from(vec![Some(true), None, None, Some(true), Some(false)]),
            ),
            (
                Operator::RegexNotMatch,
                BooleanArray::from(vec![Some(false), None, None, Some(true), Some(true)]),
            ),
            (
                Operator::RegexNotIMatch,
                BooleanArray::from(vec![
                    Some(false),
                    None,
                    None,
                    Some(false),
                    Some(true),
                ]),
            ),
        ] {
            let dictionary_result = evaluate(&dictionary_schema, &dictionary, op)?;
            let utf8_result = evaluate(&utf8_schema, &utf8, op)?;

            assert_eq!(dictionary_result.as_ref(), &expected);
            assert_eq!(&dictionary_result, &utf8_result);
        }

        Ok(())
    }

    #[test]
    fn regex_mismatched_array_types_error() -> Result<()> {
        // The analyzer coerces both operands of a regex operator to a common
        // string type, but an expression that bypasses it (e.g. constructed
        // directly) must return an error instead of panicking
        // (https://github.com/apache/datafusion/issues/22886)
        let schema = Schema::new(vec![
            Field::new("a", DataType::Utf8View, true),
            Field::new("b", DataType::Utf8, true),
        ]);
        let a = Arc::new(StringViewArray::from(vec!["user auth failed"])) as ArrayRef;
        let b = Arc::new(StringArray::from(vec!["(auth|login)"])) as ArrayRef;

        // construct the expression directly, without coercion
        let expr = binary(
            col("a", &schema)?,
            Operator::RegexMatch,
            col("b", &schema)?,
            &schema,
        )?;
        let batch = RecordBatch::try_new(Arc::new(schema), vec![a, b])?;
        let err = expr.evaluate(&batch).unwrap_err();
        assert_contains!(err.to_string(), "failed to downcast array");

        Ok(())
    }

    #[test]
    fn or_with_nulls_op() -> Result<()> {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Boolean, true),
            Field::new("b", DataType::Boolean, true),
        ]);
        let a = Arc::new(BooleanArray::from(vec![
            Some(true),
            Some(false),
            None,
            Some(true),
            Some(false),
            None,
            Some(true),
            Some(false),
            None,
        ])) as ArrayRef;
        let b = Arc::new(BooleanArray::from(vec![
            Some(true),
            Some(true),
            Some(true),
            Some(false),
            Some(false),
            Some(false),
            None,
            None,
            None,
        ])) as ArrayRef;

        let expected = BooleanArray::from(vec![
            Some(true),
            Some(true),
            Some(true),
            Some(true),
            Some(false),
            None,
            Some(true),
            None,
            None,
        ]);
        apply_logic_op(&Arc::new(schema), &a, &b, Operator::Or, expected)?;

        Ok(())
    }

    /// Returns (schema, a: BooleanArray, b: BooleanArray) with all possible inputs
    ///
    /// a: [true, true, true,  NULL, NULL, NULL,  false, false, false]
    /// b: [true, NULL, false, true, NULL, false, true,  NULL,  false]
    fn bool_test_arrays() -> (SchemaRef, ArrayRef, ArrayRef) {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Boolean, true),
            Field::new("b", DataType::Boolean, true),
        ]);
        let a: BooleanArray = [
            Some(true),
            Some(true),
            Some(true),
            None,
            None,
            None,
            Some(false),
            Some(false),
            Some(false),
        ]
        .iter()
        .collect();
        let b: BooleanArray = [
            Some(true),
            None,
            Some(false),
            Some(true),
            None,
            Some(false),
            Some(true),
            None,
            Some(false),
        ]
        .iter()
        .collect();
        (Arc::new(schema), Arc::new(a), Arc::new(b))
    }

    /// Returns (schema, BooleanArray) with [true, NULL, false]
    fn scalar_bool_test_array() -> (SchemaRef, ArrayRef) {
        let schema = Schema::new(vec![Field::new("a", DataType::Boolean, true)]);
        let a: BooleanArray = [Some(true), None, Some(false)].iter().collect();
        (Arc::new(schema), Arc::new(a))
    }

    #[test]
    fn eq_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(true),
            None,
            Some(false),
            None,
            None,
            None,
            Some(false),
            None,
            Some(true),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::Eq, expected).unwrap();
    }

    #[test]
    fn eq_op_bool_scalar() {
        let (schema, a) = scalar_bool_test_array();
        let expected = [Some(true), None, Some(false)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(true),
            &a,
            Operator::Eq,
            &expected,
        )
        .unwrap();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(true),
            Operator::Eq,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(true)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(false),
            &a,
            Operator::Eq,
            &expected,
        )
        .unwrap();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(false),
            Operator::Eq,
            &expected,
        )
        .unwrap();
    }

    #[test]
    fn neq_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(false),
            None,
            Some(true),
            None,
            None,
            None,
            Some(true),
            None,
            Some(false),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::NotEq, expected).unwrap();
    }

    #[test]
    fn neq_op_bool_scalar() {
        let (schema, a) = scalar_bool_test_array();
        let expected = [Some(false), None, Some(true)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(true),
            &a,
            Operator::NotEq,
            &expected,
        )
        .unwrap();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(true),
            Operator::NotEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(false)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(false),
            &a,
            Operator::NotEq,
            &expected,
        )
        .unwrap();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(false),
            Operator::NotEq,
            &expected,
        )
        .unwrap();
    }

    #[test]
    fn lt_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(false),
            None,
            Some(false),
            None,
            None,
            None,
            Some(true),
            None,
            Some(false),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::Lt, expected).unwrap();
    }

    #[test]
    fn lt_op_bool_scalar() {
        let (schema, a) = scalar_bool_test_array();
        let expected = [Some(false), None, Some(false)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(true),
            &a,
            Operator::Lt,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(true)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(true),
            Operator::Lt,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(false)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(false),
            &a,
            Operator::Lt,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(false)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(false),
            Operator::Lt,
            &expected,
        )
        .unwrap();
    }

    #[test]
    fn lt_eq_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(true),
            None,
            Some(false),
            None,
            None,
            None,
            Some(true),
            None,
            Some(true),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::LtEq, expected).unwrap();
    }

    #[test]
    fn lt_eq_op_bool_scalar() {
        let (schema, a) = scalar_bool_test_array();
        let expected = [Some(true), None, Some(false)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(true),
            &a,
            Operator::LtEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(true)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(true),
            Operator::LtEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(true)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(false),
            &a,
            Operator::LtEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(true)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(false),
            Operator::LtEq,
            &expected,
        )
        .unwrap();
    }

    #[test]
    fn gt_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(false),
            None,
            Some(true),
            None,
            None,
            None,
            Some(false),
            None,
            Some(false),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::Gt, expected).unwrap();
    }

    #[test]
    fn gt_op_bool_scalar() {
        let (schema, a) = scalar_bool_test_array();
        let expected = [Some(false), None, Some(true)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(true),
            &a,
            Operator::Gt,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(false)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(true),
            Operator::Gt,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(false)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(false),
            &a,
            Operator::Gt,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(false)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(false),
            Operator::Gt,
            &expected,
        )
        .unwrap();
    }

    #[test]
    fn gt_eq_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(true),
            None,
            Some(true),
            None,
            None,
            None,
            Some(false),
            None,
            Some(true),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::GtEq, expected).unwrap();
    }

    #[test]
    fn gt_eq_op_bool_scalar() {
        let (schema, a) = scalar_bool_test_array();
        let expected = [Some(true), None, Some(true)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(true),
            &a,
            Operator::GtEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(false)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(true),
            Operator::GtEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(false), None, Some(true)].iter().collect();
        apply_logic_op_scalar_arr(
            &schema,
            &ScalarValue::from(false),
            &a,
            Operator::GtEq,
            &expected,
        )
        .unwrap();

        let expected = [Some(true), None, Some(true)].iter().collect();
        apply_logic_op_arr_scalar(
            &schema,
            &a,
            &ScalarValue::from(false),
            Operator::GtEq,
            &expected,
        )
        .unwrap();
    }

    #[test]
    fn is_distinct_from_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(false),
            Some(true),
            Some(true),
            Some(true),
            Some(false),
            Some(true),
            Some(true),
            Some(true),
            Some(false),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::IsDistinctFrom, expected).unwrap();
    }

    #[test]
    fn is_not_distinct_from_op_bool() {
        let (schema, a, b) = bool_test_arrays();
        let expected = [
            Some(true),
            Some(false),
            Some(false),
            Some(false),
            Some(true),
            Some(false),
            Some(false),
            Some(false),
            Some(true),
        ]
        .iter()
        .collect();
        apply_logic_op(&schema, &a, &b, Operator::IsNotDistinctFrom, expected).unwrap();
    }

    #[test]
    fn relatively_deeply_nested() {
        // Reproducer for https://github.com/apache/datafusion/issues/419

        // where even relatively shallow binary expressions overflowed
        // the stack in debug builds

        let input: Vec<_> = vec![1, 2, 3, 4, 5].into_iter().map(Some).collect();
        let a: Int32Array = input.iter().collect();

        let batch = RecordBatch::try_from_iter(vec![("a", Arc::new(a) as _)]).unwrap();
        let schema = batch.schema();

        // build a left deep tree ((((a + a) + a) + a ....
        let tree_depth: i32 = 100;
        let expr = (0..tree_depth)
            .map(|_| col("a", schema.as_ref()).unwrap())
            .reduce(|l, r| binary(l, Operator::Plus, r, &schema).unwrap())
            .unwrap();

        let result = expr
            .evaluate(&batch)
            .expect("evaluation")
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");

        let expected: Int32Array = input
            .into_iter()
            .map(|i| i.map(|i| i * tree_depth))
            .collect();
        assert_eq!(result.as_ref(), &expected);
    }

    fn create_decimal_array(
        array: &[Option<i128>],
        precision: u8,
        scale: i8,
    ) -> Decimal128Array {
        let mut decimal_builder = Decimal128Builder::with_capacity(array.len());
        for value in array.iter().copied() {
            decimal_builder.append_option(value)
        }
        decimal_builder
            .finish()
            .with_precision_and_scale(precision, scale)
            .unwrap()
    }

    #[test]
    fn comparison_dict_decimal_scalar_expr_test() -> Result<()> {
        // scalar of decimal compare with dictionary decimal array
        let value_i128 = 123;
        let decimal_scalar = ScalarValue::Dictionary(
            Box::new(DataType::Int8),
            Box::new(ScalarValue::Decimal128(Some(value_i128), 25, 3)),
        );
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::Dictionary(
                Box::new(DataType::Int8),
                Box::new(DataType::Decimal128(25, 3)),
            ),
            true,
        )]));
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value_i128),
                None,
                Some(value_i128 - 1),
                Some(value_i128 + 1),
            ],
            25,
            3,
        ));

        let keys = Int8Array::from(vec![Some(0), None, Some(2), Some(3)]);
        let dictionary =
            Arc::new(DictionaryArray::try_new(keys, decimal_array)?) as ArrayRef;

        // array = scalar
        apply_logic_op_arr_scalar(
            &schema,
            &dictionary,
            &decimal_scalar,
            Operator::Eq,
            &BooleanArray::from(vec![Some(true), None, Some(false), Some(false)]),
        )
        .unwrap();
        // array != scalar
        apply_logic_op_arr_scalar(
            &schema,
            &dictionary,
            &decimal_scalar,
            Operator::NotEq,
            &BooleanArray::from(vec![Some(false), None, Some(true), Some(true)]),
        )
        .unwrap();
        //  array < scalar
        apply_logic_op_arr_scalar(
            &schema,
            &dictionary,
            &decimal_scalar,
            Operator::Lt,
            &BooleanArray::from(vec![Some(false), None, Some(true), Some(false)]),
        )
        .unwrap();

        //  array <= scalar
        apply_logic_op_arr_scalar(
            &schema,
            &dictionary,
            &decimal_scalar,
            Operator::LtEq,
            &BooleanArray::from(vec![Some(true), None, Some(true), Some(false)]),
        )
        .unwrap();
        // array > scalar
        apply_logic_op_arr_scalar(
            &schema,
            &dictionary,
            &decimal_scalar,
            Operator::Gt,
            &BooleanArray::from(vec![Some(false), None, Some(false), Some(true)]),
        )
        .unwrap();

        // array >= scalar
        apply_logic_op_arr_scalar(
            &schema,
            &dictionary,
            &decimal_scalar,
            Operator::GtEq,
            &BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]),
        )
        .unwrap();

        Ok(())
    }

    #[test]
    fn comparison_decimal_expr_test() -> Result<()> {
        // scalar of decimal compare with decimal array
        let value_i128 = 123;
        let decimal_scalar = ScalarValue::Decimal128(Some(value_i128), 25, 3);
        let schema = Arc::new(Schema::new(vec![Field::new(
            "a",
            DataType::Decimal128(25, 3),
            true,
        )]));
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value_i128),
                None,
                Some(value_i128 - 1),
                Some(value_i128 + 1),
            ],
            25,
            3,
        )) as ArrayRef;
        // array = scalar
        apply_logic_op_arr_scalar(
            &schema,
            &decimal_array,
            &decimal_scalar,
            Operator::Eq,
            &BooleanArray::from(vec![Some(true), None, Some(false), Some(false)]),
        )
        .unwrap();
        // array != scalar
        apply_logic_op_arr_scalar(
            &schema,
            &decimal_array,
            &decimal_scalar,
            Operator::NotEq,
            &BooleanArray::from(vec![Some(false), None, Some(true), Some(true)]),
        )
        .unwrap();
        //  array < scalar
        apply_logic_op_arr_scalar(
            &schema,
            &decimal_array,
            &decimal_scalar,
            Operator::Lt,
            &BooleanArray::from(vec![Some(false), None, Some(true), Some(false)]),
        )
        .unwrap();

        //  array <= scalar
        apply_logic_op_arr_scalar(
            &schema,
            &decimal_array,
            &decimal_scalar,
            Operator::LtEq,
            &BooleanArray::from(vec![Some(true), None, Some(true), Some(false)]),
        )
        .unwrap();
        // array > scalar
        apply_logic_op_arr_scalar(
            &schema,
            &decimal_array,
            &decimal_scalar,
            Operator::Gt,
            &BooleanArray::from(vec![Some(false), None, Some(false), Some(true)]),
        )
        .unwrap();

        // array >= scalar
        apply_logic_op_arr_scalar(
            &schema,
            &decimal_array,
            &decimal_scalar,
            Operator::GtEq,
            &BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]),
        )
        .unwrap();

        // scalar of different data type with decimal array
        let decimal_scalar = ScalarValue::Decimal128(Some(123_456), 10, 3);
        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, true)]));
        // scalar == array
        apply_logic_op_scalar_arr(
            &schema,
            &decimal_scalar,
            &(Arc::new(Int64Array::from(vec![Some(124), None])) as ArrayRef),
            Operator::Eq,
            &BooleanArray::from(vec![Some(false), None]),
        )
        .unwrap();

        // array != scalar
        apply_logic_op_arr_scalar(
            &schema,
            &(Arc::new(Int64Array::from(vec![Some(123), None, Some(1)])) as ArrayRef),
            &decimal_scalar,
            Operator::NotEq,
            &BooleanArray::from(vec![Some(true), None, Some(true)]),
        )
        .unwrap();

        // array < scalar
        apply_logic_op_arr_scalar(
            &schema,
            &(Arc::new(Int64Array::from(vec![Some(123), None, Some(124)])) as ArrayRef),
            &decimal_scalar,
            Operator::Lt,
            &BooleanArray::from(vec![Some(true), None, Some(false)]),
        )
        .unwrap();

        // array > scalar
        apply_logic_op_arr_scalar(
            &schema,
            &(Arc::new(Int64Array::from(vec![Some(123), None, Some(124)])) as ArrayRef),
            &decimal_scalar,
            Operator::Gt,
            &BooleanArray::from(vec![Some(false), None, Some(true)]),
        )
        .unwrap();

        let schema =
            Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, true)]));
        // array == scalar
        apply_logic_op_arr_scalar(
            &schema,
            &(Arc::new(Float64Array::from(vec![Some(123.456), None, Some(123.457)]))
                as ArrayRef),
            &decimal_scalar,
            Operator::Eq,
            &BooleanArray::from(vec![Some(true), None, Some(false)]),
        )
        .unwrap();

        // array <= scalar
        apply_logic_op_arr_scalar(
            &schema,
            &(Arc::new(Float64Array::from(vec![
                Some(123.456),
                None,
                Some(123.457),
                Some(123.45),
            ])) as ArrayRef),
            &decimal_scalar,
            Operator::LtEq,
            &BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]),
        )
        .unwrap();
        // array >= scalar
        apply_logic_op_arr_scalar(
            &schema,
            &(Arc::new(Float64Array::from(vec![
                Some(123.456),
                None,
                Some(123.457),
                Some(123.45),
            ])) as ArrayRef),
            &decimal_scalar,
            Operator::GtEq,
            &BooleanArray::from(vec![Some(true), None, Some(true), Some(false)]),
        )
        .unwrap();

        let value: i128 = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[Some(value), None, Some(value - 1), Some(value + 1)],
            10,
            0,
        )) as ArrayRef;

        // comparison array op for decimal array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Decimal128(10, 0), true),
            Field::new("b", DataType::Decimal128(10, 0), true),
        ]));
        let right_decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value - 1),
                Some(value),
                Some(value + 1),
                Some(value + 1),
            ],
            10,
            0,
        )) as ArrayRef;

        apply_logic_op(
            &schema,
            &decimal_array,
            &right_decimal_array,
            Operator::Eq,
            BooleanArray::from(vec![Some(false), None, Some(false), Some(true)]),
        )
        .unwrap();

        apply_logic_op(
            &schema,
            &decimal_array,
            &right_decimal_array,
            Operator::NotEq,
            BooleanArray::from(vec![Some(true), None, Some(true), Some(false)]),
        )
        .unwrap();

        apply_logic_op(
            &schema,
            &decimal_array,
            &right_decimal_array,
            Operator::Lt,
            BooleanArray::from(vec![Some(false), None, Some(true), Some(false)]),
        )
        .unwrap();

        apply_logic_op(
            &schema,
            &decimal_array,
            &right_decimal_array,
            Operator::LtEq,
            BooleanArray::from(vec![Some(false), None, Some(true), Some(true)]),
        )
        .unwrap();

        apply_logic_op(
            &schema,
            &decimal_array,
            &right_decimal_array,
            Operator::Gt,
            BooleanArray::from(vec![Some(true), None, Some(false), Some(false)]),
        )
        .unwrap();

        apply_logic_op(
            &schema,
            &decimal_array,
            &right_decimal_array,
            Operator::GtEq,
            BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]),
        )
        .unwrap();

        // compare decimal array with other array type
        let value: i64 = 123;
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int64, true),
            Field::new("b", DataType::Decimal128(10, 0), true),
        ]));

        let int64_array = Arc::new(Int64Array::from(vec![
            Some(value),
            Some(value - 1),
            Some(value),
            Some(value + 1),
        ])) as ArrayRef;

        // eq: int64array == decimal array
        apply_logic_op(
            &schema,
            &int64_array,
            &decimal_array,
            Operator::Eq,
            BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]),
        )
        .unwrap();
        // neq: int64array != decimal array
        apply_logic_op(
            &schema,
            &int64_array,
            &decimal_array,
            Operator::NotEq,
            BooleanArray::from(vec![Some(false), None, Some(true), Some(false)]),
        )
        .unwrap();

        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Float64, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));

        let value: i128 = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value), // 1.23
                None,
                Some(value - 1), // 1.22
                Some(value + 1), // 1.24
            ],
            10,
            2,
        )) as ArrayRef;
        let float64_array = Arc::new(Float64Array::from(vec![
            Some(1.23),
            Some(1.22),
            Some(1.23),
            Some(1.24),
        ])) as ArrayRef;
        // lt: float64array < decimal array
        apply_logic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Lt,
            BooleanArray::from(vec![Some(false), None, Some(false), Some(false)]),
        )
        .unwrap();
        // lt_eq: float64array <= decimal array
        apply_logic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::LtEq,
            BooleanArray::from(vec![Some(true), None, Some(false), Some(true)]),
        )
        .unwrap();
        // gt: float64array > decimal array
        apply_logic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Gt,
            BooleanArray::from(vec![Some(false), None, Some(true), Some(false)]),
        )
        .unwrap();
        apply_logic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::GtEq,
            BooleanArray::from(vec![Some(true), None, Some(true), Some(true)]),
        )
        .unwrap();
        // is distinct: float64array is distinct decimal array
        // TODO: now we do not refactor the `is distinct or is not distinct` rule of coercion.
        // traced by https://github.com/apache/datafusion/issues/1590
        // the decimal array will be casted to float64array
        apply_logic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::IsDistinctFrom,
            BooleanArray::from(vec![Some(false), Some(true), Some(true), Some(false)]),
        )
        .unwrap();
        // is not distinct
        apply_logic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::IsNotDistinctFrom,
            BooleanArray::from(vec![Some(true), Some(false), Some(false), Some(true)]),
        )
        .unwrap();

        Ok(())
    }

    fn apply_decimal_arithmetic_op(
        schema: &SchemaRef,
        left: &ArrayRef,
        right: &ArrayRef,
        op: Operator,
        expected: ArrayRef,
    ) -> Result<()> {
        let arithmetic_op = binary_op(col("a", schema)?, op, col("b", schema)?, schema)?;
        let data: Vec<ArrayRef> = vec![Arc::clone(left), Arc::clone(right)];
        let batch = RecordBatch::try_new(Arc::clone(schema), data)?;
        let result = arithmetic_op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");

        assert_eq!(result.as_ref(), expected.as_ref());
        Ok(())
    }

    #[test]
    fn arithmetic_decimal_expr_test() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let value: i128 = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value), // 1.23
                None,
                Some(value - 1), // 1.22
                Some(value + 1), // 1.24
            ],
            10,
            2,
        )) as ArrayRef;
        let int32_array = Arc::new(Int32Array::from(vec![
            Some(123),
            Some(122),
            Some(123),
            Some(124),
        ])) as ArrayRef;

        // add: Int32array add decimal array
        let expect = Arc::new(create_decimal_array(
            &[Some(12423), None, Some(12422), Some(12524)],
            13,
            2,
        )) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &int32_array,
            &decimal_array,
            Operator::Plus,
            expect,
        )
        .unwrap();

        // subtract: decimal array subtract int32 array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Decimal128(10, 2), true),
            Field::new("b", DataType::Int32, true),
        ]));
        let expect = Arc::new(create_decimal_array(
            &[Some(-12177), None, Some(-12178), Some(-12276)],
            13,
            2,
        )) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &decimal_array,
            &int32_array,
            Operator::Minus,
            expect,
        )
        .unwrap();

        // multiply: decimal array multiply int32 array
        let expect = Arc::new(create_decimal_array(
            &[Some(15129), None, Some(15006), Some(15376)],
            21,
            2,
        )) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &decimal_array,
            &int32_array,
            Operator::Multiply,
            expect,
        )
        .unwrap();

        // divide: int32 array divide decimal array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let expect = Arc::new(create_decimal_array(
            &[Some(1000000), None, Some(1008196), Some(1000000)],
            16,
            4,
        )) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &int32_array,
            &decimal_array,
            Operator::Divide,
            expect,
        )
        .unwrap();

        // modulus: int32 array modulus decimal array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let expect = Arc::new(create_decimal_array(
            &[Some(000), None, Some(100), Some(000)],
            10,
            2,
        )) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &int32_array,
            &decimal_array,
            Operator::Modulo,
            expect,
        )
        .unwrap();

        Ok(())
    }

    #[test]
    fn arithmetic_decimal_float_expr_test() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Float64, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let value: i128 = 123;
        let decimal_array = Arc::new(create_decimal_array(
            &[
                Some(value), // 1.23
                None,
                Some(value - 1), // 1.22
                Some(value + 1), // 1.24
            ],
            10,
            2,
        )) as ArrayRef;
        let float64_array = Arc::new(Float64Array::from(vec![
            Some(123.0),
            Some(122.0),
            Some(123.0),
            Some(124.0),
        ])) as ArrayRef;

        // add: float64 array add decimal array
        let expect = Arc::new(Float64Array::from(vec![
            Some(124.23),
            None,
            Some(124.22),
            Some(125.24),
        ])) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Plus,
            expect,
        )
        .unwrap();

        // subtract: decimal array subtract float64 array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Float64, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let expect = Arc::new(Float64Array::from(vec![
            Some(121.77),
            None,
            Some(121.78),
            Some(122.76),
        ])) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Minus,
            expect,
        )
        .unwrap();

        // multiply: decimal array multiply float64 array
        let expect = Arc::new(Float64Array::from(vec![
            Some(151.29),
            None,
            Some(150.06),
            Some(153.76),
        ])) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Multiply,
            expect,
        )
        .unwrap();

        // divide: float64 array divide decimal array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Float64, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let expect = Arc::new(Float64Array::from(vec![
            Some(100.0),
            None,
            Some(100.81967213114754),
            Some(100.0),
        ])) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Divide,
            expect,
        )
        .unwrap();

        // modulus: float64 array modulus decimal array
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Float64, true),
            Field::new("b", DataType::Decimal128(10, 2), true),
        ]));
        let expect = Arc::new(Float64Array::from(vec![
            Some(1.7763568394002505e-15),
            None,
            Some(1.0000000000000027),
            Some(8.881784197001252e-16),
        ])) as ArrayRef;
        apply_decimal_arithmetic_op(
            &schema,
            &float64_array,
            &decimal_array,
            Operator::Modulo,
            expect,
        )
        .unwrap();

        Ok(())
    }

    #[test]
    fn arithmetic_divide_zero() -> Result<()> {
        // other data type
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, true),
            Field::new("b", DataType::Int32, true),
        ]));
        let a = Arc::new(Int32Array::from(vec![100]));
        let b = Arc::new(Int32Array::from(vec![0]));

        let err = apply_arithmetic::<Int32Type>(
            schema,
            vec![a, b],
            Operator::Divide,
            Int32Array::from(vec![Some(4), Some(8), Some(16), Some(32), Some(64)]),
        )
        .unwrap_err();

        let _expected = plan_datafusion_err!("Divide by zero");

        assert!(matches!(err, ref _expected), "{err}");

        // decimal
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Decimal128(25, 3), true),
            Field::new("b", DataType::Decimal128(25, 3), true),
        ]));
        let left_decimal_array = Arc::new(create_decimal_array(&[Some(1234567)], 25, 3));
        let right_decimal_array = Arc::new(create_decimal_array(&[Some(0)], 25, 3));

        let err = apply_arithmetic::<Decimal128Type>(
            schema,
            vec![left_decimal_array, right_decimal_array],
            Operator::Divide,
            create_decimal_array(
                &[Some(12345670000000000000000000000000000), None],
                38,
                29,
            ),
        )
        .unwrap_err();

        assert!(matches!(err, ref _expected), "{err}");

        Ok(())
    }

    #[test]
    fn bitwise_array_test() -> Result<()> {
        let left = Arc::new(Int32Array::from(vec![Some(12), None, Some(11)])) as ArrayRef;
        let right =
            Arc::new(Int32Array::from(vec![Some(1), Some(3), Some(7)])) as ArrayRef;
        let mut result = bitwise_and_dyn(Arc::clone(&left), Arc::clone(&right))?;
        let expected = Int32Array::from(vec![Some(0), None, Some(3)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_or_dyn(Arc::clone(&left), Arc::clone(&right))?;
        let expected = Int32Array::from(vec![Some(13), None, Some(15)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_xor_dyn(Arc::clone(&left), Arc::clone(&right))?;
        let expected = Int32Array::from(vec![Some(13), None, Some(12)]);
        assert_eq!(result.as_ref(), &expected);

        let left =
            Arc::new(UInt32Array::from(vec![Some(12), None, Some(11)])) as ArrayRef;
        let right =
            Arc::new(UInt32Array::from(vec![Some(1), Some(3), Some(7)])) as ArrayRef;
        let mut result = bitwise_and_dyn(Arc::clone(&left), Arc::clone(&right))?;
        let expected = UInt32Array::from(vec![Some(0), None, Some(3)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_or_dyn(Arc::clone(&left), Arc::clone(&right))?;
        let expected = UInt32Array::from(vec![Some(13), None, Some(15)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_xor_dyn(Arc::clone(&left), Arc::clone(&right))?;
        let expected = UInt32Array::from(vec![Some(13), None, Some(12)]);
        assert_eq!(result.as_ref(), &expected);

        Ok(())
    }

    #[test]
    fn bitwise_shift_array_test() -> Result<()> {
        let input = Arc::new(Int32Array::from(vec![Some(2), None, Some(10)])) as ArrayRef;
        let modules =
            Arc::new(Int32Array::from(vec![Some(2), Some(4), Some(8)])) as ArrayRef;
        let mut result =
            bitwise_shift_left_dyn(Arc::clone(&input), Arc::clone(&modules))?;

        let expected = Int32Array::from(vec![Some(8), None, Some(2560)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_shift_right_dyn(Arc::clone(&result), Arc::clone(&modules))?;
        assert_eq!(result.as_ref(), &input);

        let input =
            Arc::new(UInt32Array::from(vec![Some(2), None, Some(10)])) as ArrayRef;
        let modules =
            Arc::new(UInt32Array::from(vec![Some(2), Some(4), Some(8)])) as ArrayRef;
        let mut result =
            bitwise_shift_left_dyn(Arc::clone(&input), Arc::clone(&modules))?;

        let expected = UInt32Array::from(vec![Some(8), None, Some(2560)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_shift_right_dyn(Arc::clone(&result), Arc::clone(&modules))?;
        assert_eq!(result.as_ref(), &input);
        Ok(())
    }

    #[test]
    fn bitwise_shift_array_overflow_test() -> Result<()> {
        let input = Arc::new(Int32Array::from(vec![Some(2)])) as ArrayRef;
        let modules = Arc::new(Int32Array::from(vec![Some(100)])) as ArrayRef;
        let result = bitwise_shift_left_dyn(Arc::clone(&input), Arc::clone(&modules))?;

        let expected = Int32Array::from(vec![Some(32)]);
        assert_eq!(result.as_ref(), &expected);

        let input = Arc::new(UInt32Array::from(vec![Some(2)])) as ArrayRef;
        let modules = Arc::new(UInt32Array::from(vec![Some(100)])) as ArrayRef;
        let result = bitwise_shift_left_dyn(Arc::clone(&input), Arc::clone(&modules))?;

        let expected = UInt32Array::from(vec![Some(32)]);
        assert_eq!(result.as_ref(), &expected);
        Ok(())
    }

    #[test]
    fn bitwise_scalar_test() -> Result<()> {
        let left = Arc::new(Int32Array::from(vec![Some(12), None, Some(11)])) as ArrayRef;
        let right = ScalarValue::from(3i32);
        let mut result = bitwise_and_dyn_scalar(&left, right.clone()).unwrap()?;
        let expected = Int32Array::from(vec![Some(0), None, Some(3)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_or_dyn_scalar(&left, right.clone()).unwrap()?;
        let expected = Int32Array::from(vec![Some(15), None, Some(11)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_xor_dyn_scalar(&left, right).unwrap()?;
        let expected = Int32Array::from(vec![Some(15), None, Some(8)]);
        assert_eq!(result.as_ref(), &expected);

        let left =
            Arc::new(UInt32Array::from(vec![Some(12), None, Some(11)])) as ArrayRef;
        let right = ScalarValue::from(3u32);
        let mut result = bitwise_and_dyn_scalar(&left, right.clone()).unwrap()?;
        let expected = UInt32Array::from(vec![Some(0), None, Some(3)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_or_dyn_scalar(&left, right.clone()).unwrap()?;
        let expected = UInt32Array::from(vec![Some(15), None, Some(11)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_xor_dyn_scalar(&left, right).unwrap()?;
        let expected = UInt32Array::from(vec![Some(15), None, Some(8)]);
        assert_eq!(result.as_ref(), &expected);
        Ok(())
    }

    #[test]
    fn bitwise_shift_scalar_test() -> Result<()> {
        let input = Arc::new(Int32Array::from(vec![Some(2), None, Some(4)])) as ArrayRef;
        let module = ScalarValue::from(10i32);
        let mut result =
            bitwise_shift_left_dyn_scalar(&input, module.clone()).unwrap()?;

        let expected = Int32Array::from(vec![Some(2048), None, Some(4096)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_shift_right_dyn_scalar(&result, module).unwrap()?;
        assert_eq!(result.as_ref(), &input);

        let input = Arc::new(UInt32Array::from(vec![Some(2), None, Some(4)])) as ArrayRef;
        let module = ScalarValue::from(10u32);
        let mut result =
            bitwise_shift_left_dyn_scalar(&input, module.clone()).unwrap()?;

        let expected = UInt32Array::from(vec![Some(2048), None, Some(4096)]);
        assert_eq!(result.as_ref(), &expected);

        result = bitwise_shift_right_dyn_scalar(&result, module).unwrap()?;
        assert_eq!(result.as_ref(), &input);
        Ok(())
    }

    #[test]
    fn test_display_and_or_combo() {
        let expr = BinaryExpr::new(
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(1)),
                Operator::And,
                lit(ScalarValue::from(2)),
            )),
            Operator::And,
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(3)),
                Operator::And,
                lit(ScalarValue::from(4)),
            )),
        );
        assert_eq!(expr.to_string(), "1 AND 2 AND 3 AND 4");

        let expr = BinaryExpr::new(
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(1)),
                Operator::Or,
                lit(ScalarValue::from(2)),
            )),
            Operator::Or,
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(3)),
                Operator::Or,
                lit(ScalarValue::from(4)),
            )),
        );
        assert_eq!(expr.to_string(), "1 OR 2 OR 3 OR 4");

        let expr = BinaryExpr::new(
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(1)),
                Operator::And,
                lit(ScalarValue::from(2)),
            )),
            Operator::Or,
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(3)),
                Operator::And,
                lit(ScalarValue::from(4)),
            )),
        );
        assert_eq!(expr.to_string(), "1 AND 2 OR 3 AND 4");

        let expr = BinaryExpr::new(
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(1)),
                Operator::Or,
                lit(ScalarValue::from(2)),
            )),
            Operator::And,
            Arc::new(BinaryExpr::new(
                lit(ScalarValue::from(3)),
                Operator::Or,
                lit(ScalarValue::from(4)),
            )),
        );
        assert_eq!(expr.to_string(), "(1 OR 2) AND (3 OR 4)");
    }

    #[test]
    fn test_to_result_type_array() {
        let values = Arc::new(Int32Array::from(vec![1, 2, 3, 4]));
        let keys = Int8Array::from(vec![Some(0), None, Some(2), Some(3)]);
        let dictionary =
            Arc::new(DictionaryArray::try_new(keys, values).unwrap()) as ArrayRef;

        // Casting Dictionary to Int32
        let casted = to_result_type_array(
            &Operator::Plus,
            Arc::clone(&dictionary),
            &DataType::Int32,
        )
        .unwrap();
        assert_eq!(
            &casted,
            &(Arc::new(Int32Array::from(vec![Some(1), None, Some(3), Some(4)]))
                as ArrayRef)
        );

        // Array has same datatype as result type, no casting
        let casted = to_result_type_array(
            &Operator::Plus,
            Arc::clone(&dictionary),
            dictionary.data_type(),
        )
        .unwrap();
        assert_eq!(&casted, &dictionary);

        // Not numerical operator, no casting
        let casted = to_result_type_array(
            &Operator::Eq,
            Arc::clone(&dictionary),
            &DataType::Int32,
        )
        .unwrap();
        assert_eq!(&casted, &dictionary);
    }

    #[test]
    fn test_add_with_overflow() -> Result<()> {
        // create test data
        let l = Arc::new(Int32Array::from(vec![1, i32::MAX]));
        let r = Arc::new(Int32Array::from(vec![2, 1]));
        let schema = Arc::new(Schema::new(vec![
            Field::new("l", DataType::Int32, false),
            Field::new("r", DataType::Int32, false),
        ]));
        let batch = RecordBatch::try_new(schema, vec![l, r])?;

        // create expression
        let expr = BinaryExpr::new(
            Arc::new(Column::new("l", 0)),
            Operator::Plus,
            Arc::new(Column::new("r", 1)),
        )
        .with_fail_on_overflow(true);

        // evaluate expression
        let result = expr.evaluate(&batch);
        assert!(
            result
                .err()
                .unwrap()
                .to_string()
                .contains("Overflow happened on: 2147483647 + 1")
        );
        Ok(())
    }

    #[test]
    fn test_subtract_with_overflow() -> Result<()> {
        // create test data
        let l = Arc::new(Int32Array::from(vec![1, i32::MIN]));
        let r = Arc::new(Int32Array::from(vec![2, 1]));
        let schema = Arc::new(Schema::new(vec![
            Field::new("l", DataType::Int32, false),
            Field::new("r", DataType::Int32, false),
        ]));
        let batch = RecordBatch::try_new(schema, vec![l, r])?;

        // create expression
        let expr = BinaryExpr::new(
            Arc::new(Column::new("l", 0)),
            Operator::Minus,
            Arc::new(Column::new("r", 1)),
        )
        .with_fail_on_overflow(true);

        // evaluate expression
        let result = expr.evaluate(&batch);
        assert!(
            result
                .err()
                .unwrap()
                .to_string()
                .contains("Overflow happened on: -2147483648 - 1")
        );
        Ok(())
    }

    #[test]
    fn test_mul_with_overflow() -> Result<()> {
        // create test data
        let l = Arc::new(Int32Array::from(vec![1, i32::MAX]));
        let r = Arc::new(Int32Array::from(vec![2, 2]));
        let schema = Arc::new(Schema::new(vec![
            Field::new("l", DataType::Int32, false),
            Field::new("r", DataType::Int32, false),
        ]));
        let batch = RecordBatch::try_new(schema, vec![l, r])?;

        // create expression
        let expr = BinaryExpr::new(
            Arc::new(Column::new("l", 0)),
            Operator::Multiply,
            Arc::new(Column::new("r", 1)),
        )
        .with_fail_on_overflow(true);

        // evaluate expression
        let result = expr.evaluate(&batch);
        assert!(
            result
                .err()
                .unwrap()
                .to_string()
                .contains("Overflow happened on: 2147483647 * 2")
        );
        Ok(())
    }

    /// Test helper for SIMILAR TO binary operation
    fn apply_similar_to(
        schema: &SchemaRef,
        va: Vec<&str>,
        vb: Vec<&str>,
        negated: bool,
        case_insensitive: bool,
        expected: &BooleanArray,
    ) -> Result<()> {
        let a = StringArray::from(va);
        let b = StringArray::from(vb);
        let op = similar_to(
            negated,
            case_insensitive,
            col("a", schema)?,
            col("b", schema)?,
        )?;
        let batch =
            RecordBatch::try_new(Arc::clone(schema), vec![Arc::new(a), Arc::new(b)])?;
        let result = op
            .evaluate(&batch)?
            .into_array(batch.num_rows())
            .expect("Failed to convert to array");
        assert_eq!(result.as_ref(), expected);

        Ok(())
    }

    #[test]
    fn test_similar_to() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Utf8, false),
            Field::new("b", DataType::Utf8, false),
        ]));

        let expected = [Some(true), Some(false)].iter().collect();
        // case-sensitive
        apply_similar_to(
            &schema,
            vec!["hello world", "Hello World"],
            vec!["hello.*", "hello.*"],
            false,
            false,
            &expected,
        )
        .unwrap();
        // case-insensitive
        apply_similar_to(
            &schema,
            vec!["hello world", "bye"],
            vec!["hello.*", "hello.*"],
            false,
            true,
            &expected,
        )
        .unwrap();
    }

    pub fn binary_expr(
        left: Arc<dyn PhysicalExpr>,
        op: Operator,
        right: Arc<dyn PhysicalExpr>,
        schema: &Schema,
    ) -> Result<BinaryExpr> {
        Ok(binary_op(left, op, right, schema)?
            .downcast_ref::<BinaryExpr>()
            .unwrap()
            .clone())
    }

    /// Test for Uniform-Uniform, Unknown-Uniform, Uniform-Unknown and Unknown-Unknown evaluation.
    #[test]
    #[expect(deprecated)]
    fn test_evaluate_statistics_combination_of_range_holders() -> Result<()> {
        let schema = &Schema::new(vec![Field::new("a", DataType::Float64, false)]);
        let a = Arc::new(Column::new("a", 0)) as _;
        let b = lit(ScalarValue::from(12.0));

        let left_interval = Interval::make(Some(0.0), Some(12.0))?;
        let right_interval = Interval::make(Some(12.0), Some(36.0))?;
        let (left_mean, right_mean) = (ScalarValue::from(6.0), ScalarValue::from(24.0));
        let (left_med, right_med) = (ScalarValue::from(6.0), ScalarValue::from(24.0));

        for children in [
            vec![
                &Distribution::new_uniform(left_interval.clone())?,
                &Distribution::new_uniform(right_interval.clone())?,
            ],
            vec![
                &Distribution::new_generic(
                    left_mean.clone(),
                    left_med.clone(),
                    ScalarValue::Float64(None),
                    left_interval.clone(),
                )?,
                &Distribution::new_uniform(right_interval.clone())?,
            ],
            vec![
                &Distribution::new_uniform(right_interval.clone())?,
                &Distribution::new_generic(
                    right_mean.clone(),
                    right_med.clone(),
                    ScalarValue::Float64(None),
                    right_interval.clone(),
                )?,
            ],
            vec![
                &Distribution::new_generic(
                    left_mean.clone(),
                    left_med.clone(),
                    ScalarValue::Float64(None),
                    left_interval.clone(),
                )?,
                &Distribution::new_generic(
                    right_mean.clone(),
                    right_med.clone(),
                    ScalarValue::Float64(None),
                    right_interval.clone(),
                )?,
            ],
        ] {
            let ops = vec![
                Operator::Plus,
                Operator::Minus,
                Operator::Multiply,
                Operator::Divide,
            ];

            for op in ops {
                let expr = binary_expr(Arc::clone(&a), op, Arc::clone(&b), schema)?;
                assert_eq!(
                    expr.evaluate_statistics(&children)?,
                    new_generic_from_binary_op(&op, children[0], children[1])?
                );
            }
        }
        Ok(())
    }

    #[test]
    #[expect(deprecated)]
    fn test_evaluate_statistics_bernoulli() -> Result<()> {
        let schema = &Schema::new(vec![
            Field::new("a", DataType::Int64, false),
            Field::new("b", DataType::Int64, false),
        ]);
        let a = Arc::new(Column::new("a", 0)) as _;
        let b = Arc::new(Column::new("b", 1)) as _;
        let eq = Arc::new(binary_expr(
            Arc::clone(&a),
            Operator::Eq,
            Arc::clone(&b),
            schema,
        )?);
        let neq = Arc::new(binary_expr(a, Operator::NotEq, b, schema)?);

        let left_stat = &Distribution::new_uniform(Interval::make(Some(0), Some(7))?)?;
        let right_stat = &Distribution::new_uniform(Interval::make(Some(4), Some(11))?)?;

        // Intervals: [0, 7], [4, 11].
        // The intersection is [4, 7], so the probability of equality is 4 / 64 = 1 / 16.
        assert_eq!(
            eq.evaluate_statistics(&[left_stat, right_stat])?,
            Distribution::new_bernoulli(ScalarValue::from(1.0 / 16.0))?
        );

        // The probability of being distinct is 1 - 1 / 16 = 15 / 16.
        assert_eq!(
            neq.evaluate_statistics(&[left_stat, right_stat])?,
            Distribution::new_bernoulli(ScalarValue::from(15.0 / 16.0))?
        );

        Ok(())
    }

    #[test]
    #[expect(deprecated)]
    fn test_propagate_statistics_combination_of_range_holders_arithmetic() -> Result<()> {
        let schema = &Schema::new(vec![Field::new("a", DataType::Float64, false)]);
        let a = Arc::new(Column::new("a", 0)) as _;
        let b = lit(ScalarValue::from(12.0));

        let left_interval = Interval::make(Some(0.0), Some(12.0))?;
        let right_interval = Interval::make(Some(12.0), Some(36.0))?;

        let parent = Distribution::new_uniform(Interval::make(Some(-432.), Some(432.))?)?;
        let children = vec![
            vec![
                Distribution::new_uniform(left_interval.clone())?,
                Distribution::new_uniform(right_interval.clone())?,
            ],
            vec![
                Distribution::new_generic(
                    ScalarValue::from(6.),
                    ScalarValue::from(6.),
                    ScalarValue::Float64(None),
                    left_interval.clone(),
                )?,
                Distribution::new_uniform(right_interval.clone())?,
            ],
            vec![
                Distribution::new_uniform(left_interval.clone())?,
                Distribution::new_generic(
                    ScalarValue::from(12.),
                    ScalarValue::from(12.),
                    ScalarValue::Float64(None),
                    right_interval.clone(),
                )?,
            ],
            vec![
                Distribution::new_generic(
                    ScalarValue::from(6.),
                    ScalarValue::from(6.),
                    ScalarValue::Float64(None),
                    left_interval.clone(),
                )?,
                Distribution::new_generic(
                    ScalarValue::from(12.),
                    ScalarValue::from(12.),
                    ScalarValue::Float64(None),
                    right_interval.clone(),
                )?,
            ],
        ];

        let ops = vec![
            Operator::Plus,
            Operator::Minus,
            Operator::Multiply,
            Operator::Divide,
        ];

        for child_view in children {
            let child_refs = child_view.iter().collect::<Vec<_>>();
            for op in &ops {
                let expr = binary_expr(Arc::clone(&a), *op, Arc::clone(&b), schema)?;
                assert_eq!(
                    expr.propagate_statistics(&parent, child_refs.as_slice())?,
                    Some(child_view.clone())
                );
            }
        }
        Ok(())
    }

    #[test]
    #[expect(deprecated)]
    fn test_propagate_statistics_combination_of_range_holders_comparison() -> Result<()> {
        let schema = &Schema::new(vec![Field::new("a", DataType::Float64, false)]);
        let a = Arc::new(Column::new("a", 0)) as _;
        let b = lit(ScalarValue::from(12.0));

        let left_interval = Interval::make(Some(0.0), Some(12.0))?;
        let right_interval = Interval::make(Some(6.0), Some(18.0))?;

        let one = ScalarValue::from(1.0);
        let parent = Distribution::new_bernoulli(one)?;
        let children = vec![
            vec![
                Distribution::new_uniform(left_interval.clone())?,
                Distribution::new_uniform(right_interval.clone())?,
            ],
            vec![
                Distribution::new_generic(
                    ScalarValue::from(6.),
                    ScalarValue::from(6.),
                    ScalarValue::Float64(None),
                    left_interval.clone(),
                )?,
                Distribution::new_uniform(right_interval.clone())?,
            ],
            vec![
                Distribution::new_uniform(left_interval.clone())?,
                Distribution::new_generic(
                    ScalarValue::from(12.),
                    ScalarValue::from(12.),
                    ScalarValue::Float64(None),
                    right_interval.clone(),
                )?,
            ],
            vec![
                Distribution::new_generic(
                    ScalarValue::from(6.),
                    ScalarValue::from(6.),
                    ScalarValue::Float64(None),
                    left_interval.clone(),
                )?,
                Distribution::new_generic(
                    ScalarValue::from(12.),
                    ScalarValue::from(12.),
                    ScalarValue::Float64(None),
                    right_interval.clone(),
                )?,
            ],
        ];

        let ops = vec![
            Operator::Eq,
            Operator::Gt,
            Operator::GtEq,
            Operator::Lt,
            Operator::LtEq,
        ];

        for child_view in children {
            let child_refs = child_view.iter().collect::<Vec<_>>();
            for op in &ops {
                let expr = binary_expr(Arc::clone(&a), *op, Arc::clone(&b), schema)?;
                assert!(
                    expr.propagate_statistics(&parent, child_refs.as_slice())?
                        .is_some()
                );
            }
        }

        Ok(())
    }

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

        // Test basic binary expressions
        let simple_expr = binary_expr(
            col("a", &schema)?,
            Operator::Plus,
            col("b", &schema)?,
            &schema,
        )?;
        let display_string = simple_expr.to_string();
        assert_eq!(display_string, "a@0 + b@1");
        let sql_string = fmt_sql(&simple_expr).to_string();
        assert_eq!(sql_string, "a + b");

        // Test nested expressions with different operator precedence
        let nested_expr = binary_expr(
            Arc::new(binary_expr(
                col("a", &schema)?,
                Operator::Plus,
                col("b", &schema)?,
                &schema,
            )?),
            Operator::Multiply,
            col("b", &schema)?,
            &schema,
        )?;
        let display_string = nested_expr.to_string();
        assert_eq!(display_string, "(a@0 + b@1) * b@1");
        let sql_string = fmt_sql(&nested_expr).to_string();
        assert_eq!(sql_string, "(a + b) * b");

        // Test nested expressions with same operator precedence
        let nested_same_prec = binary_expr(
            Arc::new(binary_expr(
                col("a", &schema)?,
                Operator::Plus,
                col("b", &schema)?,
                &schema,
            )?),
            Operator::Plus,
            col("b", &schema)?,
            &schema,
        )?;
        let display_string = nested_same_prec.to_string();
        assert_eq!(display_string, "a@0 + b@1 + b@1");
        let sql_string = fmt_sql(&nested_same_prec).to_string();
        assert_eq!(sql_string, "a + b + b");

        // Test with literals
        let lit_expr = binary_expr(
            col("a", &schema)?,
            Operator::Eq,
            lit(ScalarValue::Int32(Some(42))),
            &schema,
        )?;
        let display_string = lit_expr.to_string();
        assert_eq!(display_string, "a@0 = 42");
        let sql_string = fmt_sql(&lit_expr).to_string();
        assert_eq!(sql_string, "a = 42");

        Ok(())
    }

    #[test]
    fn test_check_short_circuit() {
        // Test with non-nullable arrays
        let schema = Arc::new(Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]));
        let a_array = Int32Array::from(vec![1, 3, 4, 5, 6]);
        let b_array = Int32Array::from(vec![1, 2, 3, 4, 5]);
        let batch = RecordBatch::try_new(
            Arc::clone(&schema),
            vec![Arc::new(a_array), Arc::new(b_array)],
        )
        .unwrap();

        // op: AND left: all false
        let left_expr = logical2physical(&logical_col("a").eq(expr_lit(2)), &schema);
        let left_value = left_expr.evaluate(&batch).unwrap();
        assert!(matches!(
            check_short_circuit(&left_value, &Operator::And),
            ShortCircuitStrategy::ReturnLeft
        ));

        // op: AND left: not all false
        let left_expr = logical2physical(&logical_col("a").eq(expr_lit(3)), &schema);
        let left_value = left_expr.evaluate(&batch).unwrap();
        let ColumnarValue::Array(array) = &left_value else {
            panic!("Expected ColumnarValue::Array");
        };
        let ShortCircuitStrategy::PreSelection { mask, fill_value } =
            check_short_circuit(&left_value, &Operator::And)
        else {
            panic!("Expected ShortCircuitStrategy::PreSelection");
        };
        // For AND, the mask selects the rows where the LHS is true and the
        // unselected rows are filled with `false`.
        assert!(!fill_value);
        let expected_boolean_arr: Vec<_> =
            as_boolean_array(array).unwrap().iter().collect();
        let boolean_arr: Vec<_> = mask.iter().collect();
        assert_eq!(expected_boolean_arr, boolean_arr);

        // op: OR left: all true
        let left_expr = logical2physical(&logical_col("a").gt(expr_lit(0)), &schema);
        let left_value = left_expr.evaluate(&batch).unwrap();
        assert!(matches!(
            check_short_circuit(&left_value, &Operator::Or),
            ShortCircuitStrategy::ReturnLeft
        ));

        // 20% false: OR can pre-select the false rows.
        let left_expr: Arc<dyn PhysicalExpr> =
            logical2physical(&logical_col("a").gt(expr_lit(2)), &schema);
        let left_value = left_expr.evaluate(&batch).unwrap();
        let ColumnarValue::Array(array) = &left_value else {
            panic!("Expected ColumnarValue::Array");
        };
        let ShortCircuitStrategy::PreSelection { mask, fill_value } =
            check_short_circuit(&left_value, &Operator::Or)
        else {
            panic!("Expected ShortCircuitStrategy::PreSelection");
        };
        // For OR, the mask selects the rows where the LHS is false (the negation
        // of the LHS) and the unselected rows are filled with `true`.
        assert!(fill_value);
        let negated_lhs: Vec<_> = as_boolean_array(array)
            .unwrap()
            .iter()
            .map(|v| v.map(|b| !b))
            .collect();
        let boolean_arr: Vec<_> = mask.iter().collect();
        assert_eq!(negated_lhs, boolean_arr);

        // 60% false: OR falls back to normal evaluation.
        let left_expr: Arc<dyn PhysicalExpr> =
            logical2physical(&logical_col("a").gt(expr_lit(4)), &schema);
        let left_value = left_expr.evaluate(&batch).unwrap();
        assert!(matches!(
            check_short_circuit(&left_value, &Operator::Or),
            ShortCircuitStrategy::None
        ));

        // Test with nullable arrays and null values
        let schema_nullable = Arc::new(Schema::new(vec![
            Field::new("c", DataType::Boolean, true),
            Field::new("d", DataType::Boolean, true),
        ]));

        // Create arrays with null values
        let c_array = Arc::new(BooleanArray::from(vec![
            Some(true),
            Some(false),
            None,
            Some(true),
            None,
        ])) as ArrayRef;
        let d_array = Arc::new(BooleanArray::from(vec![
            Some(false),
            Some(true),
            Some(false),
            None,
            Some(true),
        ])) as ArrayRef;

        let batch_nullable = RecordBatch::try_new(
            Arc::clone(&schema_nullable),
            vec![Arc::clone(&c_array), Arc::clone(&d_array)],
        )
        .unwrap();

        // Case: Mixed values with nulls - shouldn't short-circuit for AND
        let mixed_nulls = logical2physical(&logical_col("c"), &schema_nullable);
        let mixed_nulls_value = mixed_nulls.evaluate(&batch_nullable).unwrap();
        assert!(matches!(
            check_short_circuit(&mixed_nulls_value, &Operator::And),
            ShortCircuitStrategy::None
        ));

        // Case: Mixed values with nulls - shouldn't short-circuit for OR
        assert!(matches!(
            check_short_circuit(&mixed_nulls_value, &Operator::Or),
            ShortCircuitStrategy::None
        ));

        // Test with all nulls
        let all_nulls = Arc::new(BooleanArray::from(vec![None, None, None])) as ArrayRef;
        let null_batch = RecordBatch::try_new(
            Arc::new(Schema::new(vec![Field::new("e", DataType::Boolean, true)])),
            vec![all_nulls],
        )
        .unwrap();

        let null_expr = logical2physical(&logical_col("e"), &null_batch.schema());
        let null_value = null_expr.evaluate(&null_batch).unwrap();

        // All nulls shouldn't short-circuit for AND or OR
        assert!(matches!(
            check_short_circuit(&null_value, &Operator::And),
            ShortCircuitStrategy::None
        ));
        assert!(matches!(
            check_short_circuit(&null_value, &Operator::Or),
            ShortCircuitStrategy::None
        ));

        // Test with scalar values
        // Scalar true
        let scalar_true = ColumnarValue::Scalar(ScalarValue::Boolean(Some(true)));
        assert!(matches!(
            check_short_circuit(&scalar_true, &Operator::Or),
            ShortCircuitStrategy::ReturnLeft
        )); // Should short-circuit OR
        assert!(matches!(
            check_short_circuit(&scalar_true, &Operator::And),
            ShortCircuitStrategy::ReturnRight
        )); // Should return the RHS for AND

        // Scalar false
        let scalar_false = ColumnarValue::Scalar(ScalarValue::Boolean(Some(false)));
        assert!(matches!(
            check_short_circuit(&scalar_false, &Operator::And),
            ShortCircuitStrategy::ReturnLeft
        )); // Should short-circuit AND
        assert!(matches!(
            check_short_circuit(&scalar_false, &Operator::Or),
            ShortCircuitStrategy::ReturnRight
        )); // Should return the RHS for OR

        // Scalar null
        let scalar_null = ColumnarValue::Scalar(ScalarValue::Boolean(None));
        assert!(matches!(
            check_short_circuit(&scalar_null, &Operator::And),
            ShortCircuitStrategy::None
        ));
        assert!(matches!(
            check_short_circuit(&scalar_null, &Operator::Or),
            ShortCircuitStrategy::None
        ));
    }

    /// Test for [pre_selection_scatter].
    ///
    /// `check_short_circuit` only calls this helper with a non-empty,
    /// non-null mask that is neither all true nor all false.
    #[test]
    fn test_pre_selection_scatter() {
        fn create_bool_array(bools: Vec<bool>) -> BooleanArray {
            BooleanArray::from(bools.into_iter().map(Some).collect::<Vec<_>>())
        }
        // Test sparse left with interleaved true/false
        {
            // Left: [T, F, T, F, T]
            // Right: [F, T, F] (values for 3 true positions)
            let left = create_bool_array(vec![true, false, true, false, true]);
            let right = create_bool_array(vec![false, true, false]);

            let result = pre_selection_scatter(&left, Some(&right), false).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            let expected = create_bool_array(vec![false, false, true, false, false]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
        // Test multiple consecutive true blocks
        {
            // Left: [F, T, T, F, T, T, T]
            // Right: [T, F, F, T, F]
            let left =
                create_bool_array(vec![false, true, true, false, true, true, true]);
            let right = create_bool_array(vec![true, false, false, true, false]);

            let result = pre_selection_scatter(&left, Some(&right), false).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            let expected =
                create_bool_array(vec![false, true, false, false, false, true, false]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
        // Test single true at first position
        {
            // Left: [T, F, F]
            // Right: [F]
            let left = create_bool_array(vec![true, false, false]);
            let right = create_bool_array(vec![false]);

            let result = pre_selection_scatter(&left, Some(&right), false).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            let expected = create_bool_array(vec![false, false, false]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
        // Test single true at last position
        {
            // Left: [F, F, T]
            // Right: [F]
            let left = create_bool_array(vec![false, false, true]);
            let right = create_bool_array(vec![false]);

            let result = pre_selection_scatter(&left, Some(&right), false).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            let expected = create_bool_array(vec![false, false, false]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
        // Test nulls in right array
        {
            // Left: [F, T, F, T]
            // Right: [None, Some(false)] (with null at first position)
            let left = create_bool_array(vec![false, true, false, true]);
            let right = BooleanArray::from(vec![None, Some(false)]);

            let result = pre_selection_scatter(&left, Some(&right), false).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            let expected = BooleanArray::from(vec![
                Some(false),
                None, // null from right
                Some(false),
                Some(false),
            ]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
        // OR semantics: selected rows take the RHS, unselected rows become true.
        {
            // Selection (LHS false rows): [T, F, T, F, T]
            // Right (RHS on those rows): [F, T, F]
            let left = create_bool_array(vec![true, false, true, false, true]);
            let right = create_bool_array(vec![false, true, false]);

            let result = pre_selection_scatter(&left, Some(&right), true).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            // selected rows take the RHS value; unselected rows are `true`
            let expected = create_bool_array(vec![false, true, true, true, false]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
        // OR semantics with nulls in the right array.
        {
            // Selection (LHS false rows): [F, T, F, T]
            // Right: [None, Some(false)]
            let left = create_bool_array(vec![false, true, false, true]);
            let right = BooleanArray::from(vec![None, Some(false)]);

            let result = pre_selection_scatter(&left, Some(&right), true).unwrap();
            let result_arr = result.into_array(left.len()).unwrap();

            let expected = BooleanArray::from(vec![
                Some(true), // unselected => true
                None,       // null from right
                Some(true), // unselected => true
                Some(false),
            ]);
            assert_eq!(&expected, result_arr.as_boolean());
        }
    }

    #[test]
    fn test_and_true_preselection_returns_lhs() {
        let schema =
            Arc::new(Schema::new(vec![Field::new("c", DataType::Boolean, false)]));
        let c_array = Arc::new(BooleanArray::from(vec![false, true, false, false, false]))
            as ArrayRef;
        let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::clone(&c_array)])
            .unwrap();

        let expr = logical2physical(&logical_col("c").and(expr_lit(true)), &schema);

        let result = expr.evaluate(&batch).unwrap();
        let ColumnarValue::Array(result_arr) = result else {
            panic!("Expected ColumnarValue::Array");
        };

        let expected: Vec<_> = c_array.as_boolean().iter().collect();
        let actual: Vec<_> = result_arr.as_boolean().iter().collect();
        assert_eq!(
            expected, actual,
            "AND with TRUE must equal LHS even with PreSelection"
        );
    }

    #[test]
    fn test_or_false_preselection_returns_lhs() {
        // `c OR false` over a mostly-true `c` triggers OR pre-selection; the
        // result must equal `c`.
        let schema =
            Arc::new(Schema::new(vec![Field::new("c", DataType::Boolean, false)]));
        let c_array =
            Arc::new(BooleanArray::from(vec![true, false, true, true, true])) as ArrayRef;
        let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::clone(&c_array)])
            .unwrap();

        let expr = logical2physical(&logical_col("c").or(expr_lit(false)), &schema);

        let result = expr.evaluate(&batch).unwrap();
        let ColumnarValue::Array(result_arr) = result else {
            panic!("Expected ColumnarValue::Array");
        };

        let expected: Vec<_> = c_array.as_boolean().iter().collect();
        let actual: Vec<_> = result_arr.as_boolean().iter().collect();
        assert_eq!(
            expected, actual,
            "OR with FALSE must equal LHS even with PreSelection"
        );
    }

    #[test]
    fn test_or_preselection_matches_kleene() {
        // The OR pre-selection path must match full-batch Kleene OR.
        use arrow::compute::kernels::boolean::or_kleene;

        let schema = Arc::new(Schema::new(vec![
            Field::new("c", DataType::Boolean, true),
            Field::new("d", DataType::Boolean, true),
        ]));

        // `c` is mostly true (2/10 false => 20% <= threshold) so OR pre-selects.
        let c = BooleanArray::from(vec![
            true, true, false, true, true, true, true, false, true, true,
        ]);

        let d_cases = vec![
            // Mixed RHS with nulls exercises scatter and null copy.
            BooleanArray::from(vec![
                Some(false),
                Some(true),
                Some(true),
                Some(false),
                Some(false),
                Some(true),
                Some(false),
                None,
                Some(true),
                None,
            ]),
            // RHS true on selected rows exercises the uniform-fill path.
            BooleanArray::from(vec![Some(true); 10]),
            // RHS false on selected rows exercises the return-LHS path.
            BooleanArray::from(vec![Some(false); 10]),
        ];

        for d in d_cases {
            let batch = RecordBatch::try_new(
                Arc::clone(&schema),
                vec![
                    Arc::new(c.clone()) as ArrayRef,
                    Arc::new(d.clone()) as ArrayRef,
                ],
            )
            .unwrap();

            let expr = logical2physical(&logical_col("c").or(logical_col("d")), &schema);
            let result = expr.evaluate(&batch).unwrap().into_array(c.len()).unwrap();

            let expected = or_kleene(&c, &d).unwrap();
            assert_eq!(
                expected,
                *result.as_boolean(),
                "OR pre-selection must match Kleene OR for d = {d:?}"
            );
        }
    }

    #[test]
    fn test_evaluate_bounds_int32() {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Int32, false),
            Field::new("b", DataType::Int32, false),
        ]);

        let a = Arc::new(Column::new("a", 0)) as _;
        let b = Arc::new(Column::new("b", 1)) as _;

        // Test addition bounds
        let add_expr =
            binary_expr(Arc::clone(&a), Operator::Plus, Arc::clone(&b), &schema).unwrap();
        let add_bounds = add_expr
            .evaluate_bounds(&[
                &Interval::make(Some(1), Some(10)).unwrap(),
                &Interval::make(Some(5), Some(15)).unwrap(),
            ])
            .unwrap();
        assert_eq!(add_bounds, Interval::make(Some(6), Some(25)).unwrap());

        // Test subtraction bounds
        let sub_expr =
            binary_expr(Arc::clone(&a), Operator::Minus, Arc::clone(&b), &schema)
                .unwrap();
        let sub_bounds = sub_expr
            .evaluate_bounds(&[
                &Interval::make(Some(1), Some(10)).unwrap(),
                &Interval::make(Some(5), Some(15)).unwrap(),
            ])
            .unwrap();
        assert_eq!(sub_bounds, Interval::make(Some(-14), Some(5)).unwrap());

        // Test multiplication bounds
        let mul_expr =
            binary_expr(Arc::clone(&a), Operator::Multiply, Arc::clone(&b), &schema)
                .unwrap();
        let mul_bounds = mul_expr
            .evaluate_bounds(&[
                &Interval::make(Some(1), Some(10)).unwrap(),
                &Interval::make(Some(5), Some(15)).unwrap(),
            ])
            .unwrap();
        assert_eq!(mul_bounds, Interval::make(Some(5), Some(150)).unwrap());

        // Test division bounds
        let div_expr =
            binary_expr(Arc::clone(&a), Operator::Divide, Arc::clone(&b), &schema)
                .unwrap();
        let div_bounds = div_expr
            .evaluate_bounds(&[
                &Interval::make(Some(10), Some(20)).unwrap(),
                &Interval::make(Some(2), Some(5)).unwrap(),
            ])
            .unwrap();
        assert_eq!(div_bounds, Interval::make(Some(2), Some(10)).unwrap());
    }

    #[test]
    fn test_evaluate_bounds_bool() {
        let schema = Schema::new(vec![
            Field::new("a", DataType::Boolean, false),
            Field::new("b", DataType::Boolean, false),
        ]);

        let a = Arc::new(Column::new("a", 0)) as _;
        let b = Arc::new(Column::new("b", 1)) as _;

        // Test OR bounds
        let or_expr =
            binary_expr(Arc::clone(&a), Operator::Or, Arc::clone(&b), &schema).unwrap();
        let or_bounds = or_expr
            .evaluate_bounds(&[
                &Interval::make(Some(true), Some(true)).unwrap(),
                &Interval::make(Some(false), Some(false)).unwrap(),
            ])
            .unwrap();
        assert_eq!(or_bounds, Interval::make(Some(true), Some(true)).unwrap());

        // Test AND bounds
        let and_expr =
            binary_expr(Arc::clone(&a), Operator::And, Arc::clone(&b), &schema).unwrap();
        let and_bounds = and_expr
            .evaluate_bounds(&[
                &Interval::make(Some(true), Some(true)).unwrap(),
                &Interval::make(Some(false), Some(false)).unwrap(),
            ])
            .unwrap();
        assert_eq!(
            and_bounds,
            Interval::make(Some(false), Some(false)).unwrap()
        );
    }

    #[test]
    fn test_evaluate_nested_type() {
        let batch_schema = Arc::new(Schema::new(vec![
            Field::new(
                "a",
                DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))),
                true,
            ),
            Field::new(
                "b",
                DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))),
                true,
            ),
        ]));

        let mut list_builder_a = ListBuilder::new(Int32Builder::new());

        list_builder_a.append_value([Some(1)]);
        list_builder_a.append_value([Some(2)]);
        list_builder_a.append_value([]);
        list_builder_a.append_value([None]);

        let list_array_a: ArrayRef = Arc::new(list_builder_a.finish());

        let mut list_builder_b = ListBuilder::new(Int32Builder::new());

        list_builder_b.append_value([Some(1)]);
        list_builder_b.append_value([Some(2)]);
        list_builder_b.append_value([]);
        list_builder_b.append_value([None]);

        let list_array_b: ArrayRef = Arc::new(list_builder_b.finish());

        let batch =
            RecordBatch::try_new(batch_schema, vec![list_array_a, list_array_b]).unwrap();

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

        let a = Arc::new(Column::new("a", 0)) as _;
        let b = Arc::new(Column::new("b", 1)) as _;

        let eq_expr =
            binary_expr(Arc::clone(&a), Operator::Eq, Arc::clone(&b), &schema).unwrap();

        let eq_result = eq_expr.evaluate(&batch).unwrap();
        let expected =
            BooleanArray::from_iter(vec![Some(true), Some(true), Some(true), Some(true)]);
        assert_eq!(eq_result.into_array(4).unwrap().as_boolean(), &expected);
    }
}