You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/02 20:30:57 UTC
(datafusion) branch main updated: perf: evaluate AND chains as one n-ary conjunction (#25035) (#25771)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 52e964c96f perf: evaluate AND chains as one n-ary conjunction (#25035) (#25771)
52e964c96f is described below
commit 52e964c96f3cb5c38eef24abc23432b1e9c1eb2e
Author: Abhishek <16...@users.noreply.github.com>
AuthorDate: Fri Oct 2 20:21:03 2026 +0000
perf: evaluate AND chains as one n-ary conjunction (#25035) (#25771)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #25035.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
Please explain the problem you are trying to solve in terms of the
user-visible
behavior, rather than the implementation.
For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of
the
implementation. "COUNT(DISTINCT) returns wrong results when the column
contains
nulls" is the user-visible problem.
-->
In a left-deep `AND` chain, the accumulated prefix eventually drops
below the 20% pre-selection threshold, and from then on every binary
node filters the whole batch (all columns) and scatters back. Cheap
predicates on wide batches get several times slower, and the same
predicate costs very different amounts depending on how the tree is
nested.
Option B alone (one n-ary pass, filtering once) makes both shapes equal
but not fast: filtering and scattering cost more than a cheap comparison
over all rows. So this also makes the filtering decision depend on the
next conjunct.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here, but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
`BinaryExpr::evaluate` now evaluates an `AND` and the `AND`s nested
under it as one n-ary conjunction (`evaluate_conjunction`), in order:
- Before each conjunct, the batch is filtered down to the rows not yet
`false`, only when at most 20% of rows remain and the conjunct is not
cheap and infallible (`is_cheap_and_infallible`).
- `NULL` rows stay, since a later `false` still decides them.
- Evaluation stops once every row is `false`, and results are scattered
back once at the end.
The tree, `EXPLAIN`, proto and optimizer are untouched. `OR` is
unchanged; `check_short_circuit` just loses its now unused `AND`
branches. A `conjunction` benchmark is added to `binary_op.rs`.
## Things reviewers should know
1. **Overlap with #22698.** It edits the `AND` branch of
`check_short_circuit` that this PR removes, and exports
`and_rhs_evaluation`, which models the old per-node decision. We need to
agree which lands first, hence the draft.
2. **Contract.** An expensive or fallible conjunct never sees a row that
nested binary evaluation would have skipped, for any tree shape, so no
query can start failing. This relies on the cheap list only holding
things that cannot fail on a row: comparisons, `IN` lists, `NOT`, `IS
[NOT] NULL`, lossless casts (`CastExpr::check_bigger_cast`) and wrapping
integer/float `+ - *`. I assume comparisons only fail on type
mismatches, never on values. Please double check this list.
3. **Behaviour changes, same results:**
- Cheap conjuncts no longer trigger pre-selection.
- A `NULL` in the prefix no longer disables pre-selection before
expensive conjuncts.
- A non-boolean `AND` operand returns an internal error instead of
panicking.
4. **Known loss.** `predicate_eval` `selectivity_q21` is about 10%
slower (1.03 → 1.14 ms): a very selective, clustered prefix on a narrow
table, where filtering before a cheap conjunct still pays off. A
calibration sweep showed filtering first only helps when very few rows
survive (about 2 to 5%) on narrow batches, and costs up to 3.4×
otherwise, so I kept the simple rule.
5. `predicate_eval` doesn't run on `main` right now: its template has
the `result` line twice (#25037, #25032). I removed the duplicate
locally to measure and will send that one-line fix separately.
## What is the testing strategy for this PR?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
Briefly describe how this PR is tested, and point to the specific tests
you added. For example: 'This new feature is covered by the
`sqllogictest` cases added in `foo.slt`'.
If this PR does not add tests, explain why. For example, if the change
is already covered by existing tests, please mention it.
You should also check the `codecov` bot reply on this PR to confirm the
changed code is exercised.
-->
New tests in `binary.rs`:
- Randomized check against a reference Kleene `AND` over left-deep,
right-deep and balanced trees, with NULLs and literals.
- Rows seen by each conjunct, identical for every tree shape.
- NULL rows stay undecided.
- A division-by-zero conjunct only sees the rows left.
- The cheap/fallible classification and the filtering decision.
- A non-boolean operand errors instead of panicking.
The shape, NULL and non-boolean tests fail on `main`.
`cargo fmt`, `cargo clippy --all-targets --all-features -- -D warnings`
and the extended test run (`--profile ci` with `extended_tests`,
including all 524 sqllogictest files) pass: 12,159 tests. The only
failure, `sort_with_mem_limit_1_runner`, fails the same way on `main` on
my machine (an RSS limit, with no `AND` in the query).
### Benchmarks
i7-11700F, Linux 6.1, rustc 1.98.1. Micro: Criterion means, 8192-row
batches, 70% pass rate per conjunct unless noted. End to end:
`release-nonlto`, interleaved runs with a `main` vs `main` control.
`cargo bench --bench binary_op -- conjunction`, µs, left-deep /
right-deep:
| Chain | `main` | This PR |
|---|---|---|
| 8 × `c < x` | 87 / 35 | 31 / 32 |
| 8 × `c < x`, 32 unused columns | 204 / 35 | 32 / 32 |
| 16 × `c < x`, 32 unused columns | 438 / 72 | 60 / 59 |
| 8 × `CAST(c AS BIGINT) < x` | 102 / 59 | 54 / 58 |
| 8 × `c IN (...)` | 110 / 70 | 69 / 68 |
| 8 × `c + 1 < x` | 95 / 57 | 47 / 44 |
| 8 × `c < x`, then a regex | 406 / 489 | 282 / 282 |
| 4 nullable `c < x` at 50%, then a regex | 441 / 460 | 332 / 353 |
| 8 × `c < x` at 90% (never filters) | 36 / 35 | 31 / 33 |
The existing `short_circuit/and` benchmarks are unchanged or up to 8%
faster.
TPC-H (Parquet), ms:
| Query | SF1 `main` → PR | SF10 `main` → PR |
|---|---|---|
| Q6 | 49.0 → 42.0 (−14%) | 386 → 327 (−15%) |
| Q12 | 44.6 → 38.5 (−14%) | 383 → 320 (−17%) |
| Q19 | 90.5 → 82.7 (−9%) | 849 → 744 (−12%) |
These three were faster in every round. At SF1, Q16 is 4% faster (5 of 6
rounds) and every other query is within ±4%; at SF10, Q1 and Q14 are
within 2%.
`predicate_eval` (1M rows, 8 rounds, results validated): geometric mean
0.974 (`main` vs `main`: 1.005), `cardinality_q33_k16` 13.1 → 5.5 ms
(2.4×). Apart from `selectivity_q21` above, no query is more than about
5% slower, except `drift_q83`, which is just as noisy `main` vs `main`.
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
No API changes. `AND` filters get faster, and fallible predicates are
never evaluated on more rows than before.
---
datafusion/physical-expr/src/expressions/binary.rs | 847 +++++++++++++++++----
1 file changed, 678 insertions(+), 169 deletions(-)
diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs
index a0a6548518..44df100ad8 100644
--- a/datafusion/physical-expr/src/expressions/binary.rs
+++ b/datafusion/physical-expr/src/expressions/binary.rs
@@ -21,6 +21,7 @@ use crate::PhysicalExpr;
use crate::expressions::SqlSimilarToPattern;
use crate::expressions::translate_scalar;
use crate::intervals::cp_solver::{propagate_arithmetic, propagate_comparison};
+use std::borrow::Cow;
use std::cmp::Ordering;
use std::hash::Hash;
use std::sync::Arc;
@@ -28,7 +29,7 @@ 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::compute::{SlicesIterator, cast, filter, filter_record_batch};
use arrow::datatypes::*;
use arrow::error::ArrowError;
use datafusion_common::cast::as_boolean_array;
@@ -559,6 +560,10 @@ impl PhysicalExpr for BinaryExpr {
fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
use arrow::compute::kernels::numeric::*;
+ if self.op == Operator::And {
+ return self.evaluate_conjunction(batch);
+ }
+
// Evaluate left-hand side expression.
let lhs = self.left.evaluate(batch)?;
@@ -1167,19 +1172,7 @@ impl BinaryExpr {
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()
- )
- }
- }
+ | NotLikeMatch | NotILikeMatch | And => unreachable!(),
Or => {
if left_data_type == &DataType::Boolean {
Ok(boolean_op(&left, &right, or_kleene)?)
@@ -1212,69 +1205,87 @@ impl BinaryExpr {
}
}
}
+
+ /// Evaluates nested `AND`s together to avoid repeated filtering (#25035).
+ ///
+ /// When at most [`PRE_SELECTION_THRESHOLD`] of original rows remain,
+ /// filters before conjuncts that may be costly or fail, so they never see
+ /// rows nested evaluation would skip. `NULL` rows stay for a later `false`.
+ fn evaluate_conjunction(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
+ // Keep each filter mask for the final scatter.
+ let mut selections = vec![];
+ let mut input = Cow::Borrowed(batch);
+ let mut result = ColumnarValue::Scalar(ScalarValue::Boolean(Some(true)));
+
+ // Returns whether to go on to the next conjunct.
+ let mut evaluate = |conjunct: &Arc<dyn PhysicalExpr>| -> Result<bool> {
+ if let Some(undecided) = rows_to_filter_before(
+ conjunct,
+ &result,
+ batch.schema_ref(),
+ batch.num_rows(),
+ ) {
+ let array = result.to_array(input.num_rows())?;
+ // Every kept row is true when there are no NULLs.
+ result = if array.null_count() == 0 {
+ ColumnarValue::Scalar(ScalarValue::Boolean(Some(true)))
+ } else {
+ ColumnarValue::Array(filter(&array, &undecided)?)
+ };
+ input = Cow::Owned(filter_record_batch(&input, &undecided)?);
+ selections.push(undecided);
+ }
+
+ let value = conjunct.evaluate(&input)?;
+ let so_far =
+ std::mem::replace(&mut result, ColumnarValue::Scalar(ScalarValue::Null));
+ result = and_kleene_columnar(so_far, value, input.num_rows())?;
+ Ok(!is_all_false(&result))
+ };
+ if for_each_conjunct(&self.left, &mut evaluate)? {
+ for_each_conjunct(&self.right, &mut evaluate)?;
+ }
+
+ selections
+ .into_iter()
+ .rev()
+ .try_fold(result, |result, selection| unfilter(selection, result))
+ }
}
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`).
+ /// Evaluate the RHS where the LHS is `false`; fill other rows with `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
+/// Maximum share of rows still needing evaluation before filtering.
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
+/// Returns the short-circuit strategy for `OR` with a Boolean LHS.
+/// All true returns the LHS; all false returns the RHS. Sparse false rows
+/// trigger pre-selection at [`PRE_SELECTION_THRESHOLD`].
+/// `AND` uses [`BinaryExpr::evaluate_conjunction`].
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,
- };
+ if *op != Operator::Or {
+ return ShortCircuitStrategy::None;
+ }
- // Non-boolean types can't be short-circuited
+ // Only Boolean values can short-circuit.
if lhs.data_type() != DataType::Boolean {
return ShortCircuitStrategy::None;
}
match lhs {
ColumnarValue::Array(array) => {
- // Fast path for arrays - try to downcast to boolean array
+ // Check Boolean arrays directly.
if let Ok(bool_array) = as_boolean_array(array) {
- // Arrays with nulls can't be short-circuited
+ // NULLs still need the RHS.
if bool_array.null_count() > 0 {
return ShortCircuitStrategy::None;
}
@@ -1285,70 +1296,43 @@ fn check_short_circuit(lhs: &ColumnarValue, op: &Operator) -> ShortCircuitStrate
}
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 == len {
+ return ShortCircuitStrategy::ReturnLeft;
+ }
- if true_count == 0 {
- return ShortCircuitStrategy::ReturnRight;
- }
+ 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,
- };
- }
+ let false_count = len - true_count;
+ if false_count as f32 / len as f32 <= PRE_SELECTION_THRESHOLD {
+ // Only false rows need the RHS; the LHS has no NULLs here.
+ let mask = BooleanArray::new(!bool_array.values(), None);
+ return ShortCircuitStrategy::PreSelection {
+ mask,
+ fill_value: true,
+ };
}
}
}
ColumnarValue::Scalar(scalar) => {
- // Fast path for scalar values
+ // Scalars need no mask.
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;
+ return if *is_true {
+ ShortCircuitStrategy::ReturnLeft
} else {
- return ShortCircuitStrategy::ReturnRight;
- }
+ ShortCircuitStrategy::ReturnRight
+ };
}
}
}
- // If we can't short-circuit, indicate that normal evaluation should continue
+ // Evaluate both sides normally.
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`.
+/// Avoids scattering when the selected RHS rows are uniform.
+/// Returns a scalar if they match `fill_value`; otherwise returns the LHS.
fn uniform_pre_selection_result(
rhs_value: bool,
fill_value: bool,
@@ -1361,21 +1345,8 @@ fn uniform_pre_selection_result(
}
}
-/// 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`.
+/// Scatters compact RHS values where `mask` is true and fills other rows with
+/// `fill_value`. If `right_result` is `None`, selected rows become NULL.
fn pre_selection_scatter(
mask: &BooleanArray,
right_result: Option<&BooleanArray>,
@@ -1394,7 +1365,7 @@ fn pre_selection_scatter(
result_array_builder.append_n(start - last_end, fill_value);
}
- // copy values from right array for this slice
+ // Copy the selected RHS slice.
let len = end - start;
right_result
.slice(right_array_pos, len)
@@ -1417,7 +1388,7 @@ fn pre_selection_scatter(
}),
}
- // Fill any remaining positions with `fill_value`
+ // Fill the tail.
if last_end < result_len {
result_array_builder.append_n(result_len - last_end, fill_value);
}
@@ -1426,6 +1397,150 @@ fn pre_selection_scatter(
Ok(ColumnarValue::Array(Arc::new(boolean_result)))
}
+/// Selects undecided rows when some are false, at most
+/// [`PRE_SELECTION_THRESHOLD`] of the original batch remain, and evaluating
+/// `conjunct` on all rows is not cheap and infallible.
+fn rows_to_filter_before(
+ conjunct: &Arc<dyn PhysicalExpr>,
+ result: &ColumnarValue,
+ schema: &Schema,
+ num_rows: usize,
+) -> Option<BooleanArray> {
+ let ColumnarValue::Array(array) = result else {
+ return None;
+ };
+ let undecided = not_false(array.as_boolean());
+ let undecided_count = undecided.true_count();
+ // Classify last: it walks the whole conjunct.
+ (undecided_count < array.len()
+ && undecided_count as f32 / num_rows as f32 <= PRE_SELECTION_THRESHOLD
+ && !is_cheap_and_infallible(conjunct, schema))
+ .then_some(undecided)
+}
+
+/// Whether evaluating `expr` on all rows is cheap and cannot fail.
+/// Inspired by `reorder_predicates`' cheap forms, with fallible forms excluded.
+fn is_cheap_and_infallible(expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
+ use crate::expressions::{
+ CastExpr, Column, InListExpr, IsNotNullExpr, IsNullExpr, Literal, NotExpr,
+ };
+ use Operator::*;
+
+ let is_number = |expr: &Arc<dyn PhysicalExpr>| {
+ expr.data_type(schema)
+ .is_ok_and(|t| t.is_integer() || t.is_floating())
+ };
+ let is_cheap = if let Some(binary) = expr.downcast_ref::<BinaryExpr>() {
+ match binary.op {
+ Eq | NotEq | Lt | LtEq | Gt | GtEq | IsDistinctFrom | IsNotDistinctFrom
+ | And | Or => true,
+ // Wraps on overflow instead of failing.
+ Plus | Minus | Multiply => {
+ !binary.fail_on_overflow
+ && is_number(&binary.left)
+ && is_number(&binary.right)
+ }
+ _ => false,
+ }
+ } else if let Some(cast) = expr.downcast_ref::<CastExpr>() {
+ cast.expr()
+ .data_type(schema)
+ .is_ok_and(|from| cast.is_bigger_cast(&from))
+ } else {
+ expr.is::<Column>()
+ || expr.is::<Literal>()
+ || expr.is::<InListExpr>()
+ || expr.is::<NotExpr>()
+ || expr.is::<IsNullExpr>()
+ || expr.is::<IsNotNullExpr>()
+ };
+ is_cheap
+ && expr
+ .children()
+ .into_iter()
+ .all(|child| is_cheap_and_infallible(child, schema))
+}
+
+/// Calls `f` on each conjunct of the `AND` chain under `expr`, left to right,
+/// until it returns `false`. Walks the tree in place, so it doesn't allocate.
+fn for_each_conjunct(
+ expr: &Arc<dyn PhysicalExpr>,
+ f: &mut impl FnMut(&Arc<dyn PhysicalExpr>) -> Result<bool>,
+) -> Result<bool> {
+ match expr.downcast_ref::<BinaryExpr>() {
+ Some(binary) if binary.op == Operator::And => {
+ Ok(for_each_conjunct(&binary.left, f)?
+ && for_each_conjunct(&binary.right, f)?)
+ }
+ _ => f(expr),
+ }
+}
+
+/// Whether every row is already `false`.
+fn is_all_false(value: &ColumnarValue) -> bool {
+ match value {
+ ColumnarValue::Array(array) => {
+ let array = array.as_boolean();
+ array.null_count() == 0 && !array.has_true()
+ }
+ ColumnarValue::Scalar(scalar) => *scalar == ScalarValue::Boolean(Some(false)),
+ }
+}
+
+/// Scatters `result` back to all rows of `selection`; dropped rows are `false`.
+fn unfilter(selection: BooleanArray, result: ColumnarValue) -> Result<ColumnarValue> {
+ let result = result.into_array(selection.true_count())?;
+ let result = result.as_boolean();
+ if result.null_count() == 0 {
+ if !result.has_false() {
+ return Ok(ColumnarValue::Array(Arc::new(selection)));
+ }
+ if !result.has_true() {
+ return Ok(ColumnarValue::Scalar(ScalarValue::Boolean(Some(false))));
+ }
+ }
+ pre_selection_scatter(&selection, Some(result), false)
+}
+
+/// A null-free mask of the rows of `array` that are `true` or `NULL`.
+fn not_false(array: &BooleanArray) -> BooleanArray {
+ let values = match array.nulls() {
+ Some(nulls) => array.values() | &!nulls.inner(),
+ None => array.values().clone(),
+ };
+ BooleanArray::new(values, None)
+}
+
+/// Kleene `AND` of `lhs` and `rhs` over `num_rows` rows.
+fn and_kleene_columnar(
+ lhs: ColumnarValue,
+ rhs: ColumnarValue,
+ num_rows: usize,
+) -> Result<ColumnarValue> {
+ use ColumnarValue::Scalar;
+ use ScalarValue::Boolean;
+
+ if rhs.data_type() != DataType::Boolean {
+ return internal_err!("Cannot evaluate AND with a {} operand", rhs.data_type());
+ }
+ Ok(match (lhs, rhs) {
+ (Scalar(Boolean(Some(true))), other) | (other, Scalar(Boolean(Some(true)))) => {
+ other
+ }
+ (Scalar(Boolean(Some(false))), _) | (_, Scalar(Boolean(Some(false)))) => {
+ Scalar(Boolean(Some(false)))
+ }
+ (lhs, rhs) => {
+ let lhs = lhs.into_array(num_rows)?;
+ let rhs = rhs.into_array(num_rows)?;
+ ColumnarValue::Array(Arc::new(and_kleene(
+ lhs.as_boolean(),
+ rhs.as_boolean(),
+ )?))
+ }
+ })
+}
+
/// 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.
@@ -6038,33 +6153,14 @@ mod tests {
)
.unwrap();
- // op: AND left: all false
+ // `AND` uses `evaluate_conjunction` instead.
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
+ ShortCircuitStrategy::None
));
- // 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();
@@ -6133,15 +6229,9 @@ mod tests {
)
.unwrap();
- // Case: Mixed values with nulls - shouldn't short-circuit for AND
+ // Case: Mixed values with nulls - shouldn't short-circuit for OR
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
@@ -6158,11 +6248,7 @@ mod tests {
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
- ));
+ // All nulls shouldn't short-circuit for OR
assert!(matches!(
check_short_circuit(&null_value, &Operator::Or),
ShortCircuitStrategy::None
@@ -6175,17 +6261,9 @@ mod tests {
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
@@ -6193,20 +6271,13 @@ mod tests {
// 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.
+ /// Tests [`pre_selection_scatter`] with non-empty, null-free, mixed masks.
#[test]
fn test_pre_selection_scatter() {
fn create_bool_array(bools: Vec<bool>) -> BooleanArray {
@@ -6425,6 +6496,444 @@ mod tests {
}
}
+ fn left_deep_and(conjuncts: &[Arc<dyn PhysicalExpr>]) -> Arc<dyn PhysicalExpr> {
+ conjuncts
+ .iter()
+ .cloned()
+ .reduce(|l, r| Arc::new(BinaryExpr::new(l, Operator::And, r)))
+ .unwrap()
+ }
+
+ fn right_deep_and(conjuncts: &[Arc<dyn PhysicalExpr>]) -> Arc<dyn PhysicalExpr> {
+ conjuncts
+ .iter()
+ .cloned()
+ .rev()
+ .reduce(|r, l| Arc::new(BinaryExpr::new(l, Operator::And, r)))
+ .unwrap()
+ }
+
+ fn balanced_and(conjuncts: &[Arc<dyn PhysicalExpr>]) -> Arc<dyn PhysicalExpr> {
+ if let [conjunct] = conjuncts {
+ return Arc::clone(conjunct);
+ }
+ let (l, r) = conjuncts.split_at(conjuncts.len() / 2);
+ Arc::new(BinaryExpr::new(
+ balanced_and(l),
+ Operator::And,
+ balanced_and(r),
+ ))
+ }
+
+ type RowCountLog = Arc<std::sync::Mutex<Vec<(usize, usize)>>>;
+
+ /// Records rows seen by each conjunct and bypasses the cheap-expression path.
+ #[derive(Debug)]
+ struct RowCountRecorder {
+ id: usize,
+ inner: Arc<dyn PhysicalExpr>,
+ log: RowCountLog,
+ }
+
+ impl RowCountRecorder {
+ fn wrap(
+ id: usize,
+ inner: &Arc<dyn PhysicalExpr>,
+ log: &RowCountLog,
+ ) -> Arc<dyn PhysicalExpr> {
+ Arc::new(Self {
+ id,
+ inner: Arc::clone(inner),
+ log: Arc::clone(log),
+ })
+ }
+ }
+
+ impl PartialEq for RowCountRecorder {
+ fn eq(&self, other: &Self) -> bool {
+ self.id == other.id && self.inner.eq(&other.inner)
+ }
+ }
+
+ impl Eq for RowCountRecorder {}
+
+ impl Hash for RowCountRecorder {
+ fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+ self.id.hash(state);
+ self.inner.hash(state);
+ }
+ }
+
+ impl std::fmt::Display for RowCountRecorder {
+ fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+ write!(f, "{}", self.inner)
+ }
+ }
+
+ impl PhysicalExpr for RowCountRecorder {
+ fn data_type(&self, input_schema: &Schema) -> Result<DataType> {
+ self.inner.data_type(input_schema)
+ }
+
+ fn nullable(&self, input_schema: &Schema) -> Result<bool> {
+ self.inner.nullable(input_schema)
+ }
+
+ fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
+ self.log.lock().unwrap().push((self.id, batch.num_rows()));
+ self.inner.evaluate(batch)
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
+ vec![&self.inner]
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ mut children: Vec<Arc<dyn PhysicalExpr>>,
+ ) -> Result<Arc<dyn PhysicalExpr>> {
+ Ok(Self::wrap(self.id, &children.swap_remove(0), &self.log))
+ }
+
+ fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ self.inner.fmt_sql(f)
+ }
+ }
+
+ #[test]
+ fn test_conjunction_matches_kleene_and() -> Result<()> {
+ use rand::rngs::StdRng;
+ use rand::{Rng, SeedableRng};
+
+ let log = RowCountLog::default();
+ let mut rng = StdRng::seed_from_u64(25035);
+ for num_rows in [0, 1, 7, 100, 1000] {
+ for _ in 0..50 {
+ let mut fields = vec![];
+ let mut columns: Vec<ArrayRef> = vec![];
+ let mut conjuncts: Vec<Arc<dyn PhysicalExpr>> = vec![];
+ let mut expected: Option<BooleanArray> = None;
+ for id in 0..rng.random_range(2..=6) {
+ // Mostly nullable columns, sometimes a boolean literal.
+ let (conjunct, values): (Arc<dyn PhysicalExpr>, BooleanArray) = if rng
+ .random_bool(0.15)
+ {
+ let value =
+ [Some(true), Some(false), None][rng.random_range(0..3)];
+ (
+ lit(ScalarValue::Boolean(value)),
+ vec![value; num_rows].into(),
+ )
+ } else {
+ let p_true = [0.05, 0.3, 0.7, 0.95, 1.0][rng.random_range(0..5)];
+ let p_null = [0.0, 0.0, 0.1, 0.5][rng.random_range(0..4)];
+ let values = (0..num_rows)
+ .map(|_| {
+ (!rng.random_bool(p_null))
+ .then(|| rng.random_bool(p_true))
+ })
+ .collect::<BooleanArray>();
+ let index = columns.len();
+ let name = format!("c{index}");
+ fields.push(Field::new(&name, DataType::Boolean, true));
+ columns.push(Arc::new(values.clone()));
+ (Arc::new(Column::new(&name, index)), values)
+ };
+ // Opaque half the time, so filtering kicks in.
+ let conjunct = if rng.random_bool(0.5) {
+ RowCountRecorder::wrap(id, &conjunct, &log)
+ } else {
+ conjunct
+ };
+ conjuncts.push(conjunct);
+ expected = Some(match expected {
+ None => values,
+ Some(acc) => and_kleene(&acc, &values)?,
+ });
+ }
+ let expected = expected.unwrap();
+ let batch = RecordBatch::try_new_with_options(
+ Arc::new(Schema::new(fields)),
+ columns,
+ &RecordBatchOptions::new().with_row_count(Some(num_rows)),
+ )?;
+
+ for expr in [
+ left_deep_and(&conjuncts),
+ right_deep_and(&conjuncts),
+ balanced_and(&conjuncts),
+ ] {
+ let result = expr.evaluate(&batch)?.into_array(num_rows)?;
+ assert_eq!(&expected, result.as_boolean(), "{expr}");
+ }
+ }
+ }
+ // Filtering did kick in.
+ let log = log.lock().unwrap();
+ assert!(log.iter().any(|(_, rows)| (8..100).contains(rows)));
+ Ok(())
+ }
+
+ /// Compares three tree shapes and records rows seen by each conjunct.
+ fn evaluate_recorded_conjunction(
+ conjuncts: &[Arc<dyn PhysicalExpr>],
+ batch: &RecordBatch,
+ ) -> Result<(ArrayRef, Vec<usize>)> {
+ let log = RowCountLog::default();
+ let recorded: Vec<_> = conjuncts
+ .iter()
+ .enumerate()
+ .map(|(id, conjunct)| RowCountRecorder::wrap(id, conjunct, &log))
+ .collect();
+
+ let mut outcomes = vec![];
+ for expr in [
+ left_deep_and(&recorded),
+ right_deep_and(&recorded),
+ balanced_and(&recorded),
+ ] {
+ let result = expr.evaluate(batch)?.into_array(batch.num_rows())?;
+ let log = std::mem::take(&mut *log.lock().unwrap());
+ let (ids, rows): (Vec<_>, Vec<_>) = log.into_iter().unzip();
+ assert_eq!(ids, (0..conjuncts.len()).collect::<Vec<_>>(), "{expr}");
+ outcomes.push((result, rows));
+ }
+ // Tree shape must not change results or row counts.
+ assert!(outcomes.windows(2).all(|w| w[0] == w[1]), "{outcomes:?}");
+ Ok(outcomes.swap_remove(0))
+ }
+
+ #[test]
+ fn test_conjunction_filters_to_undecided_rows() -> Result<()> {
+ let schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from_iter_values(0..100))],
+ )?;
+ let id_lt =
+ |n: i32| logical2physical(&logical_col("id").lt(expr_lit(n)), &schema);
+
+ let (result, rows) = evaluate_recorded_conjunction(
+ &[
+ id_lt(50),
+ id_lt(30),
+ id_lt(20),
+ // Filter when 20% remain.
+ id_lt(10),
+ // Filter again to the undecided rows.
+ logical2physical(&logical_col("id").gt_eq(expr_lit(0)), &schema),
+ id_lt(5),
+ ],
+ &batch,
+ )?;
+ assert_eq!(rows, vec![100, 100, 100, 20, 10, 10]);
+ let expected: BooleanArray = (0..100).map(|id| Some(id < 5)).collect();
+ assert_eq!(&expected, result.as_boolean());
+ Ok(())
+ }
+
+ #[test]
+ fn test_conjunction_keeps_null_rows_undecided() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("c", DataType::Boolean, true),
+ ]));
+ // `c` is NULL for ids 0..10, true for 10..15 and false otherwise.
+ let c: BooleanArray = (0..100).map(|id| (id >= 10).then_some(id < 15)).collect();
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![Arc::new(Int32Array::from_iter_values(0..100)), Arc::new(c)],
+ )?;
+
+ let (result, rows) = evaluate_recorded_conjunction(
+ &[
+ logical2physical(&logical_col("c"), &schema),
+ logical2physical(&logical_col("id").gt_eq(expr_lit(5)), &schema),
+ ],
+ &batch,
+ )?;
+ // Keep all 10 NULL rows and the 5 true rows.
+ assert_eq!(rows, vec![100, 15]);
+ // A later false resolves the first 5 NULL rows.
+ let expected: BooleanArray = (0..100)
+ .map(|id| match id {
+ 0..5 => Some(false),
+ 5..10 => None,
+ 10..15 => Some(true),
+ _ => Some(false),
+ })
+ .collect();
+ assert_eq!(&expected, result.as_boolean());
+ Ok(())
+ }
+
+ #[test]
+ fn test_conjunction_filters_before_fallible_conjunct() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("a", DataType::Int32, false),
+ ]));
+ // `a` is only non-zero where `id < 10`.
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from_iter_values(0..100)),
+ Arc::new(Int32Array::from_iter_values(
+ (0..100).map(|id| if id < 10 { id + 1 } else { 0 }),
+ )),
+ ],
+ )?;
+ let conjuncts = [
+ logical2physical(&logical_col("id").gt_eq(expr_lit(0)), &schema),
+ logical2physical(&logical_col("id").lt(expr_lit(10)), &schema),
+ // Divides by zero on the other 90 rows.
+ logical2physical(
+ &(expr_lit(100) / logical_col("a")).gt(expr_lit(1)),
+ &schema,
+ ),
+ ];
+
+ let expected: BooleanArray = (0..100).map(|id| Some(id < 10)).collect();
+ for expr in [
+ left_deep_and(&conjuncts),
+ right_deep_and(&conjuncts),
+ balanced_and(&conjuncts),
+ ] {
+ let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
+ assert_eq!(&expected, result.as_boolean(), "{expr}");
+ }
+ Ok(())
+ }
+
+ #[test]
+ fn test_is_cheap_and_infallible() {
+ use datafusion_expr::{binary_expr, cast, not};
+
+ let schema = Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ Field::new("c", DataType::Boolean, true),
+ Field::new("s", DataType::Utf8, true),
+ ]);
+ let is_cheap = |expr: &datafusion_expr::Expr| {
+ is_cheap_and_infallible(&logical2physical(expr, &schema), &schema)
+ };
+ let (a, b, c, s) = (
+ logical_col("a"),
+ logical_col("b"),
+ logical_col("c"),
+ logical_col("s"),
+ );
+
+ assert!(is_cheap(&a.clone().lt(expr_lit(5))));
+ assert!(is_cheap(&a.clone().eq(b.clone()).and(not(c.clone()))));
+ assert!(is_cheap(
+ &c.clone().is_null().or(a.clone().gt_eq(expr_lit(1)))
+ ));
+ assert!(is_cheap(&binary_expr(
+ a.clone(),
+ Operator::IsDistinctFrom,
+ b.clone(),
+ )));
+ assert!(is_cheap(&s.clone().not_eq(expr_lit("x"))));
+ assert!(is_cheap(
+ &a.clone().in_list(vec![expr_lit(1), expr_lit(2)], false)
+ ));
+ // Widening and wrapping arithmetic cannot fail.
+ assert!(is_cheap(
+ &cast(a.clone(), DataType::Int64).lt(expr_lit(5i64))
+ ));
+ assert!(is_cheap(
+ &(a.clone() * b.clone() - expr_lit(1)).gt(a.clone())
+ ));
+
+ // Narrowing, parsing, division, and checked arithmetic can fail.
+ let checked_plus = BinaryExpr::new(
+ logical2physical(&a, &schema),
+ Operator::Plus,
+ logical2physical(&b, &schema),
+ )
+ .with_fail_on_overflow(true);
+ assert!(!is_cheap_and_infallible(
+ &(Arc::new(checked_plus) as _),
+ &schema
+ ));
+ assert!(!is_cheap(
+ &cast(a.clone(), DataType::Int8).lt(expr_lit(5i8))
+ ));
+ assert!(!is_cheap(&cast(s.clone(), DataType::Int32).gt(a.clone())));
+ assert!(!is_cheap(&(expr_lit(10) / a.clone()).gt(expr_lit(1))));
+ assert!(!is_cheap(&a.in_list(vec![expr_lit(10) / b.clone()], false)));
+ assert!(!is_cheap(&s.clone().like(expr_lit("x%"))));
+ assert!(!is_cheap(&binary_expr(
+ s,
+ Operator::RegexMatch,
+ expr_lit("x")
+ )));
+ }
+
+ #[test]
+ fn test_rows_to_filter_before() {
+ let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+ let cheap = logical2physical(&logical_col("a").lt(expr_lit(5)), &schema);
+ let fallible =
+ logical2physical(&(expr_lit(10) / logical_col("a")).gt(expr_lit(1)), &schema);
+ // 1 true and 1 NULL row out of 10.
+ let few_left = ColumnarValue::Array(Arc::new(BooleanArray::from(
+ (0..10)
+ .map(|i| (i != 1).then_some(i == 0))
+ .collect::<Vec<_>>(),
+ )));
+ let many_left = ColumnarValue::Array(Arc::new(BooleanArray::from(
+ (0..10).map(|i| Some(i < 3)).collect::<Vec<_>>(),
+ )));
+ let all_true = ColumnarValue::Array(Arc::new(BooleanArray::from(vec![true; 10])));
+
+ // Filter before fallible work when few rows remain; keep NULLs.
+ let expected = BooleanArray::from((0..10).map(|i| i < 2).collect::<Vec<_>>());
+ assert_eq!(
+ rows_to_filter_before(&fallible, &few_left, &schema, 10),
+ Some(expected)
+ );
+ assert_eq!(rows_to_filter_before(&cheap, &few_left, &schema, 10), None);
+ assert_eq!(
+ rows_to_filter_before(&fallible, &many_left, &schema, 10),
+ None
+ );
+ assert_eq!(
+ rows_to_filter_before(&fallible, &all_true, &schema, 10),
+ None
+ );
+ let scalar = ColumnarValue::Scalar(ScalarValue::Boolean(None));
+ assert_eq!(rows_to_filter_before(&fallible, &scalar, &schema, 10), None);
+ }
+
+ #[test]
+ fn test_conjunction_rejects_non_boolean_operand() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Boolean, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ Arc::clone(&schema),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(BooleanArray::from(vec![true, false])),
+ ],
+ )?;
+ let (a, b) = (col("a", &schema)?, col("b", &schema)?);
+
+ for expr in [
+ BinaryExpr::new(Arc::clone(&a), Operator::And, Arc::clone(&b)),
+ BinaryExpr::new(b, Operator::And, a),
+ ] {
+ let err = expr.evaluate(&batch).unwrap_err();
+ assert_contains!(err.to_string(), "Cannot evaluate AND with a Int32 operand");
+ }
+ Ok(())
+ }
+
#[test]
fn test_integer_interval_propagation_covers_runtime_values() {
// Enumerate small domains and both ends of Int8, including zero divisors,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org