You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JunRuiLee (via GitHub)" <gi...@apache.org> on 2026/07/03 17:34:23 UTC
[PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
JunRuiLee opened a new pull request, #448:
URL: https://github.com/apache/paimon-rust/pull/448
### Purpose
This makes predicate evaluation exact across all file formats in the Rust reader, so a consumer that returns scan batches verbatim gets exactly the rows matching the pushed-down predicate.
The motivating case is pypaimon delegating its read kernel to the Rust core: that path runs the whole scan on Rust and returns the resulting Arrow batches directly, without re-applying the predicate above the scan. Today that is only safe for Parquet — the other formats apply the predicate partially or not at all, so a filtered read silently returns rows that do not match:
- **ORC** pruned by stripe/row-group statistics only — non-matching rows inside a selected stripe survived.
- **Avro** and **Row** ignored the predicate entirely (the reader's `predicates` parameter was unused).
A silent over-read (wrong data, no error) is the worst failure mode for that delegation. After this change, every format applies the predicate exactly, closing this class of silent-wrong-read for predicate pushdown regardless of format. (Exact `LIMIT` remains the caller's responsibility, matching `ReadBuilder::with_limit`'s documented hint semantics and Paimon's Java behavior.)
### Brief change log
- Consolidate the two near-duplicate Arrow-batch predicate evaluators (one private to the Parquet reader, one in the Vortex reader) into a shared `arrow::residual` module. The Vortex copy previously fell open on `startsWith`/`endsWith`/`contains`/`like`/`between`; the shared evaluator implements them, so those operators are now evaluated exactly wherever the residual filter runs.
- Apply the shared exact residual filter after ORC stripe pruning (pruning kept as an optimization).
- Apply the residual filter in the Avro and Row readers, which previously ignored predicates.
- Parquet is unchanged in behavior (already exact; now shares the evaluator).
### Tests
- `cargo test -p paimon --lib`: all pass. Per-format exactness tests (ORC/Avro/Row) assert that reading `[10,20,30,40,50]` with `age > 25` returns exactly `[30,40,50]`, plus string-operator cases; an end-to-end test drives a `.row` file through `DataFileReader`.
- `cargo test -p paimon --lib --features vortex`: all pass (the consolidation touched the Vortex reader).
- Shared-evaluator unit tests cover the comparison/set/string/range leaves, And/Or/Not composition, and the NULL→false convention.
### API and Format
No public API or storage-format change.
### Documentation
Behavior documented inline; no separate docs update required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3522907272
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1247 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ data_type: predicate_data_type,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ // Evaluate against the predicate's declared type (the table type),
+ // not the file column's type. Under schema evolution these differ
+ // (e.g. an old INT file column for a promoted BIGINT table column):
+ // building the literal in the narrower file type could fail for an
+ // out-of-range value, incorrectly erroring when the exact answer is
+ // simply "no rows". Promotion is always widening, so cast the decoded
+ // column up to the predicate type first — then the literal is always
+ // representable and the comparison is exact.
+ let predicate_arrow_type = crate::arrow::paimon_type_to_arrow(predicate_data_type)?;
+ let mask = if column.data_type() == &predicate_arrow_type {
+ evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ } else {
+ let cast_column = arrow_cast::cast(column, &predicate_arrow_type).map_err(|e| {
+ Error::DataInvalid {
+ message: format!(
+ "Failed to cast residual column '{}' from {:?} to {:?}: {e}",
+ file_field.name(),
+ column.data_type(),
+ predicate_arrow_type
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ evaluate_exact_leaf_predicate(&cast_column, predicate_data_type, *op, literals)
+ }
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let result = evaluate_column_predicate(array, &scalar, op)?;
+ Ok(sanitize_filter_mask(result))
+ }
+ PredicateOperator::Between | PredicateOperator::NotBetween => {
+ evaluate_between_predicate(array, data_type, op, literals)
+ }
+ }
+}
+
+/// `Between` / `NotBetween` translate to `gt_eq(col, low) & lt_eq(col, high)`
+/// (and its negation). `arrow_ord::cmp` produces nullable masks: any null
+/// row makes the comparison null, so a fully-built `Between` mask preserves
+/// nulls. `NotBetween` then negates valid rows and leaves nulls null —
+/// matching SQL three-valued logic; `sanitize_filter_mask` collapses nulls
+/// into `false` to match the predicate evaluator's "NULL → false" rule.
+fn evaluate_between_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ let (Some(low), Some(high)) = (literals.first(), literals.get(1)) else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(low_scalar) = literal_scalar_for_arrow_filter(low, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(high_scalar) = literal_scalar_for_arrow_filter(high, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let lo_mask = arrow_gt_eq(array, &low_scalar)?;
+ let hi_mask = arrow_lt_eq(array, &high_scalar)?;
+ let between = arrow_arith::boolean::and_kleene(&lo_mask, &hi_mask)?;
+ let result = match op {
+ PredicateOperator::Between => between,
+ PredicateOperator::NotBetween => arrow_arith::boolean::not(&between)?,
+ _ => unreachable!(),
+ };
+ Ok(sanitize_filter_mask(result))
+}
+
+fn evaluate_set_membership_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ if literals.is_empty() {
+ return Ok(match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ });
+ }
+
+ let mut combined = match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ };
+
+ for literal in literals {
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let comparison_op = match op {
+ PredicateOperator::In => PredicateOperator::Eq,
+ PredicateOperator::NotIn => PredicateOperator::NotEq,
+ _ => unreachable!(),
+ };
+ let mask = sanitize_filter_mask(evaluate_column_predicate(array, &scalar, comparison_op)?);
+ combined = combine_filter_masks(&combined, &mask, matches!(op, PredicateOperator::In));
+ }
+
+ Ok(combined)
+}
+
+fn evaluate_column_predicate(
Review Comment:
Fixed. Binary/VarBinary comparisons (ordering and Between) now go through Paimon's signed-byte comparator (java_bytes_cmp), matching Datum::Bytes (0xFF < 0x00) instead of Arrow's unsigned order. Between delegates to the shared comparison path so it can't diverge. Regression tests added for both ordering and Between.
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1247 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ data_type: predicate_data_type,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ // Evaluate against the predicate's declared type (the table type),
+ // not the file column's type. Under schema evolution these differ
+ // (e.g. an old INT file column for a promoted BIGINT table column):
+ // building the literal in the narrower file type could fail for an
+ // out-of-range value, incorrectly erroring when the exact answer is
+ // simply "no rows". Promotion is always widening, so cast the decoded
+ // column up to the predicate type first — then the literal is always
+ // representable and the comparison is exact.
+ let predicate_arrow_type = crate::arrow::paimon_type_to_arrow(predicate_data_type)?;
+ let mask = if column.data_type() == &predicate_arrow_type {
+ evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ } else {
+ let cast_column = arrow_cast::cast(column, &predicate_arrow_type).map_err(|e| {
+ Error::DataInvalid {
+ message: format!(
+ "Failed to cast residual column '{}' from {:?} to {:?}: {e}",
+ file_field.name(),
+ column.data_type(),
+ predicate_arrow_type
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ evaluate_exact_leaf_predicate(&cast_column, predicate_data_type, *op, literals)
+ }
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let result = evaluate_column_predicate(array, &scalar, op)?;
+ Ok(sanitize_filter_mask(result))
+ }
+ PredicateOperator::Between | PredicateOperator::NotBetween => {
+ evaluate_between_predicate(array, data_type, op, literals)
+ }
+ }
+}
+
+/// `Between` / `NotBetween` translate to `gt_eq(col, low) & lt_eq(col, high)`
+/// (and its negation). `arrow_ord::cmp` produces nullable masks: any null
+/// row makes the comparison null, so a fully-built `Between` mask preserves
+/// nulls. `NotBetween` then negates valid rows and leaves nulls null —
+/// matching SQL three-valued logic; `sanitize_filter_mask` collapses nulls
+/// into `false` to match the predicate evaluator's "NULL → false" rule.
+fn evaluate_between_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ let (Some(low), Some(high)) = (literals.first(), literals.get(1)) else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(low_scalar) = literal_scalar_for_arrow_filter(low, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(high_scalar) = literal_scalar_for_arrow_filter(high, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let lo_mask = arrow_gt_eq(array, &low_scalar)?;
+ let hi_mask = arrow_lt_eq(array, &high_scalar)?;
+ let between = arrow_arith::boolean::and_kleene(&lo_mask, &hi_mask)?;
+ let result = match op {
+ PredicateOperator::Between => between,
+ PredicateOperator::NotBetween => arrow_arith::boolean::not(&between)?,
+ _ => unreachable!(),
+ };
+ Ok(sanitize_filter_mask(result))
+}
+
+fn evaluate_set_membership_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ if literals.is_empty() {
+ return Ok(match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ });
+ }
+
+ let mut combined = match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ };
+
+ for literal in literals {
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let comparison_op = match op {
+ PredicateOperator::In => PredicateOperator::Eq,
+ PredicateOperator::NotIn => PredicateOperator::NotEq,
+ _ => unreachable!(),
+ };
+ let mask = sanitize_filter_mask(evaluate_column_predicate(array, &scalar, comparison_op)?);
+ combined = combine_filter_masks(&combined, &mask, matches!(op, PredicateOperator::In));
+ }
+
+ Ok(combined)
+}
+
+fn evaluate_column_predicate(
+ column: &ArrayRef,
+ scalar: &Scalar<ArrayRef>,
+ op: PredicateOperator,
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::Eq => arrow_eq(column, scalar),
+ PredicateOperator::NotEq => arrow_neq(column, scalar),
+ PredicateOperator::Lt => arrow_lt(column, scalar),
+ PredicateOperator::LtEq => arrow_lt_eq(column, scalar),
+ PredicateOperator::Gt => arrow_gt(column, scalar),
+ PredicateOperator::GtEq => arrow_gt_eq(column, scalar),
+ PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let pattern = pattern_scalar_for_string_kernel(scalar, column.data_type())?;
+ match op {
+ PredicateOperator::StartsWith => arrow_starts_with(column, &pattern),
+ PredicateOperator::EndsWith => arrow_ends_with(column, &pattern),
+ PredicateOperator::Contains => arrow_contains(column, &pattern),
+ PredicateOperator::Like => arrow_like(column, &pattern),
+ _ => unreachable!(),
+ }
+ }
+ PredicateOperator::IsNull
+ | PredicateOperator::IsNotNull
+ | PredicateOperator::In
+ | PredicateOperator::NotIn
+ | PredicateOperator::Between
+ | PredicateOperator::NotBetween => Ok(BooleanArray::new_null(column.len())),
+ }
+}
+
+/// `arrow_string::like::*` kernels reject mismatched string types — Utf8 column
+/// against Utf8 pattern is fine, but a LargeUtf8 / Utf8View column needs a
+/// pattern of the same flavour. The shared scalar built upstream is always
+/// `StringArray` (Utf8); promote it to match the column when needed.
+fn pattern_scalar_for_string_kernel(
+ scalar: &Scalar<ArrayRef>,
+ column_type: &arrow_schema::DataType,
+) -> Result<Scalar<ArrayRef>, ArrowError> {
+ use arrow_array::{LargeStringArray, StringArray, StringViewArray};
+ use arrow_schema::DataType as ArrowDataType;
+
+ let arr = scalar.get().0;
+ let value = arr
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .and_then(|s| (s.len() == 1 && s.is_valid(0)).then(|| s.value(0).to_string()));
+ let Some(value) = value else {
+ return Ok(scalar.clone());
+ };
+ Ok(match column_type {
+ ArrowDataType::Utf8 => Scalar::new(Arc::new(StringArray::from(vec![value])) as ArrayRef),
+ ArrowDataType::LargeUtf8 => {
+ Scalar::new(Arc::new(LargeStringArray::from(vec![value])) as ArrayRef)
+ }
+ ArrowDataType::Utf8View => {
+ Scalar::new(Arc::new(StringViewArray::from(vec![value])) as ArrayRef)
+ }
+ ArrowDataType::Dictionary(_, value_type) if value_type.as_ref() == &ArrowDataType::Utf8 => {
+ Scalar::new(Arc::new(StringArray::from(vec![value])) as ArrayRef)
+ }
+ other => {
+ return Err(ArrowError::InvalidArgumentError(format!(
+ "string predicate against non-string column type {other:?}"
+ )))
+ }
+ })
+}
+
+/// Collapse `NULL` mask entries to `false`, matching the residual-filter
+/// convention. A mask with no nulls is returned unchanged.
+pub(crate) fn sanitize_filter_mask(mask: BooleanArray) -> BooleanArray {
+ if mask.null_count() == 0 {
+ return mask;
+ }
+
+ boolean_mask_from_predicate(mask.len(), |row_index| {
+ mask.is_valid(row_index) && mask.value(row_index)
+ })
+}
+
+fn combine_filter_masks(left: &BooleanArray, right: &BooleanArray, use_or: bool) -> BooleanArray {
+ debug_assert_eq!(left.len(), right.len());
+ boolean_mask_from_predicate(left.len(), |row_index| {
+ if use_or {
+ left.value(row_index) || right.value(row_index)
+ } else {
+ left.value(row_index) && right.value(row_index)
+ }
+ })
+}
+
+fn boolean_mask_from_predicate(
+ len: usize,
+ mut predicate: impl FnMut(usize) -> bool,
+) -> BooleanArray {
+ BooleanArray::from((0..len).map(&mut predicate).collect::<Vec<_>>())
+}
+
+// ---------------------------------------------------------------------------
+// Literal → Arrow scalar conversion
+// ---------------------------------------------------------------------------
+
+/// Convert a paimon [`Datum`] literal into an Arrow scalar matching
+/// `file_data_type`, or `None` when the literal / type pair is unsupported for
+/// filtering (in which case the caller falls open, keeping the row).
+pub(crate) fn literal_scalar_for_arrow_filter(
+ literal: &Datum,
+ file_data_type: &DataType,
+) -> crate::Result<Option<Scalar<ArrayRef>>> {
+ let array: ArrayRef = match file_data_type {
+ DataType::Boolean(_) => match literal {
+ Datum::Bool(value) => Arc::new(BooleanArray::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::TinyInt(_) => {
+ match integer_literal(literal).and_then(|value| i8::try_from(value).ok()) {
+ Some(value) => Arc::new(Int8Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::SmallInt(_) => {
+ match integer_literal(literal).and_then(|value| i16::try_from(value).ok()) {
+ Some(value) => Arc::new(Int16Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::Int(_) => {
+ match integer_literal(literal).and_then(|value| i32::try_from(value).ok()) {
+ Some(value) => Arc::new(Int32Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::BigInt(_) => {
+ match integer_literal(literal).and_then(|value| i64::try_from(value).ok()) {
+ Some(value) => Arc::new(Int64Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::Float(_) => match float32_literal(literal) {
+ Some(value) => Arc::new(Float32Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ },
+ DataType::Double(_) => match float64_literal(literal) {
+ Some(value) => Arc::new(Float64Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ },
+ DataType::Char(_) | DataType::VarChar(_) => match literal {
+ Datum::String(value) => Arc::new(StringArray::new_scalar(value.as_str()).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_) => match literal {
+ Datum::Bytes(value) => Arc::new(BinaryArray::new_scalar(value.as_slice()).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Date(_) => match literal {
+ Datum::Date(value) => Arc::new(Date32Array::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Time(_) => match literal {
+ Datum::Time(value) => Arc::new(Time32MillisecondArray::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Timestamp(ts) => match literal {
+ Datum::Timestamp { millis, nanos } => {
+ let Some(array) = timestamp_scalar(*millis, *nanos, ts.precision(), None)? else {
+ return Ok(None);
+ };
+ array
+ }
+ _ => return Ok(None),
+ },
+ DataType::LocalZonedTimestamp(ts) => match literal {
+ Datum::LocalZonedTimestamp { millis, nanos } => {
+ let Some(array) = timestamp_scalar(*millis, *nanos, ts.precision(), Some("UTC"))?
+ else {
+ return Ok(None);
+ };
+ array
+ }
+ _ => return Ok(None),
+ },
+ DataType::Decimal(decimal) => match literal {
Review Comment:
Fixed. Decimal value predicates are now evaluated row-wise via datum_cmp, so cross-scale comparison is by mathematical value (e.g. DECIMAL 1.0 matches a DECIMAL(_,2) column, and d > 1.05 on a DECIMAL(_,1) column is exactly d >= 1.1) — no rescale requirement and no spurious error. Tests cover the equal-value cross-scale and finer-scale-literal cases.
##########
crates/paimon/src/table/data_file_reader.rs:
##########
@@ -69,6 +69,26 @@ impl DataFileReader {
self
}
+ /// Reject projecting `_ROW_ID` alongside a data predicate. `_ROW_ID` is
+ /// assigned positionally from post-filter batch row counts, so a residual
+ /// filter that drops rows would desync it. (`_ROW_ID` predicates travel via
+ /// `row_ranges`, not `predicates`, so they do not trip this.)
+ fn reject_row_id_with_predicate(
+ read_type: &[DataField],
+ predicates: &[Predicate],
+ ) -> crate::Result<()> {
+ let projects_row_id = read_type
Review Comment:
Fixed. The guard now only rejects predicates that can actually filter rows — a constant AlwaysTrue is allowed.
Broader note: the residual filter this PR adds makes format-reader reads exact, but the data-evolution read path still passes no predicate to the reader, so index/fallback row-ranges aren't residual-filtered there. That gap is orthogonal to this PR and to #449's fallback-coverage fix; I'll follow up on it separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3522536303
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -223,7 +235,32 @@ impl FormatFileReader for ParquetFormatReader {
}
let batch_stream = batch_stream_builder.build()?;
- Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed())
+
+ if all_enforced {
+ // Fast path: the row filter enforced every predicate exactly during
+ // decode. Return the stream directly — no residual pass.
+ return Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed());
+ }
+
+ // Residual backstop: at least one predicate (`Or`/`Not`/unsupported leaf)
Review Comment:
Fixed. `ReadBuilder`/`TableRead` were pruning `And`/`Or`/`Not` out of the data predicate before the reader, so the residual pass never saw compound predicates on the public path. Rather than carry a separate pruned copy, I just stopped pruning — pushdown/stats already skip compound nodes they can't use, so the reader now gets the full predicate and filters exactly. Added a regression test through the real `ReadBuilder` -> `to_arrow` path.
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1184 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ let mask = evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
Review Comment:
Good catch, fixed. The residual now evaluates the leaf against the predicate's declared type instead of the file column's type: it casts the decoded column up to the (wider, promoted) type first, so an out-of-range value on an old narrower column yields no rows rather than an error. Added tests for the INT->BIGINT boundary (out-of-range -> 0 rows, in-range -> exact).
##########
crates/paimon/src/arrow/format/mod.rs:
##########
@@ -54,8 +54,13 @@ pub(crate) struct FilePredicates {
/// - Row range selection
#[async_trait]
pub(crate) trait FormatFileReader: Send + Sync {
- /// Read a single data file, returning a stream of RecordBatches
- /// containing only the projected columns (using names from the file's schema).
+ /// Read a single data file, returning a stream of RecordBatches containing
+ /// at least the projected columns (using names from the file's schema). A
+ /// reader MAY include extra columns it needed to scan (e.g. predicate columns
+ /// for residual filtering); the caller (`DataFileReader`) projects to the
+ /// requested output by name, so extra columns are harmless. Each batch MUST
+ /// already have the pushed-down predicate applied exactly — residual filtering
Review Comment:
Fixed — narrowed the doc. It now states exactness per format (Parquet/ORC/Avro/Row/Vortex apply the predicate exactly; Blob ignores predicates and Mosaic does stats-only pruning) instead of claiming a blanket guarantee.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3523397822
##########
crates/paimon/src/table/read_builder.rs:
##########
@@ -233,10 +232,13 @@ impl<'a> ReadBuilder<'a> {
Some(projected) => self.resolve_projected_fields(projected)?,
};
+ // Pass the FULL data predicate through (including `And`/`Or`/`Not`).
+ // Pushdown/stats skip compound nodes; the residual pass enforces the full
+ // predicate exactly. Pruning here would drop compound predicates.
Ok(TableRead::new(
self.table,
read_type,
- reader_pruning_predicates(self.filter.data_predicates.clone()),
+ self.filter.data_predicates.clone(),
Review Comment:
This still does not make the public read path exact for `data_evolution_enabled` tables. `new_read()` now preserves the full data predicate here, but `TableRead::read_with_evolution()` builds `DataEvolutionReader` without passing `self.data_predicates`, and `DataEvolutionReader` creates every `DataFileReader` with `Vec::new()`. `TableScan` also disables normal data-predicate pushdown for data-evolution tables (`pushdown_data_predicates = &[]`), so a filtered scan without a usable global index, or with only coarse row ranges, will still return non-matching rows from the data-evolution reader. This regresses the PR goal that scan batches are exactly filtered; the evolution reader needs to receive and apply the same residual predicates after column merge (or before when it is safe).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3522433673
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -223,7 +235,32 @@ impl FormatFileReader for ParquetFormatReader {
}
let batch_stream = batch_stream_builder.build()?;
- Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed())
+
+ if all_enforced {
+ // Fast path: the row filter enforced every predicate exactly during
+ // decode. Return the stream directly — no residual pass.
+ return Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed());
+ }
+
+ // Residual backstop: at least one predicate (`Or`/`Not`/unsupported leaf)
Review Comment:
Good catch — you're right, and this is the important one. `TableRead::with_filter` runs `reader_pruning_predicates()`, which drops `And`/`Or`/`Not`, so the residual backstop never sees a compound predicate on the public read path (my Parquet test exercised `read_batch_stream` directly and bypassed that). I'll thread two predicate sets through `TableRead`/`DataFileReader`: the full predicate for exact residual filtering, and a separately-pruned copy for stats/row-group pushdown only — and add a test through the real `ReadBuilder`/`TableRead` path (not the format reader directly) so this is actually covered.
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1184 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ let mask = evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
Review Comment:
Agreed — erroring here is wrong for schema-evolved files. For a promoted INT→BIGINT column read from an old INT file, a predicate value outside the old column's range should yield *no rows* for that file, not `DataInvalid`. I'll handle the numeric-range boundary (out-of-range literal → empty result / AlwaysFalse for that file) and only error on genuinely unrepresentable cases, with tests for the promotion boundaries.
##########
crates/paimon/src/arrow/format/mod.rs:
##########
@@ -54,8 +54,13 @@ pub(crate) struct FilePredicates {
/// - Row range selection
#[async_trait]
pub(crate) trait FormatFileReader: Send + Sync {
- /// Read a single data file, returning a stream of RecordBatches
- /// containing only the projected columns (using names from the file's schema).
+ /// Read a single data file, returning a stream of RecordBatches containing
+ /// at least the projected columns (using names from the file's schema). A
+ /// reader MAY include extra columns it needed to scan (e.g. predicate columns
+ /// for residual filtering); the caller (`DataFileReader`) projects to the
+ /// requested output by name, so extra columns are harmless. Each batch MUST
+ /// already have the pushed-down predicate applied exactly — residual filtering
Review Comment:
Right, the contract is stronger than Blob (ignores predicates) and Mosaic (stats pruning only). I'll narrow the doc to the formats changed here (Parquet/ORC/Avro/Row/Vortex) and note that Blob/Mosaic still rely on stats-only pruning, rather than overstating a guarantee they don't provide.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "QuakeWang (via GitHub)" <gi...@apache.org>.
QuakeWang commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3522609691
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1247 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ data_type: predicate_data_type,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ // Evaluate against the predicate's declared type (the table type),
+ // not the file column's type. Under schema evolution these differ
+ // (e.g. an old INT file column for a promoted BIGINT table column):
+ // building the literal in the narrower file type could fail for an
+ // out-of-range value, incorrectly erroring when the exact answer is
+ // simply "no rows". Promotion is always widening, so cast the decoded
+ // column up to the predicate type first — then the literal is always
+ // representable and the comparison is exact.
+ let predicate_arrow_type = crate::arrow::paimon_type_to_arrow(predicate_data_type)?;
+ let mask = if column.data_type() == &predicate_arrow_type {
+ evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ } else {
+ let cast_column = arrow_cast::cast(column, &predicate_arrow_type).map_err(|e| {
+ Error::DataInvalid {
+ message: format!(
+ "Failed to cast residual column '{}' from {:?} to {:?}: {e}",
+ file_field.name(),
+ column.data_type(),
+ predicate_arrow_type
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ evaluate_exact_leaf_predicate(&cast_column, predicate_data_type, *op, literals)
+ }
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let result = evaluate_column_predicate(array, &scalar, op)?;
+ Ok(sanitize_filter_mask(result))
+ }
+ PredicateOperator::Between | PredicateOperator::NotBetween => {
+ evaluate_between_predicate(array, data_type, op, literals)
+ }
+ }
+}
+
+/// `Between` / `NotBetween` translate to `gt_eq(col, low) & lt_eq(col, high)`
+/// (and its negation). `arrow_ord::cmp` produces nullable masks: any null
+/// row makes the comparison null, so a fully-built `Between` mask preserves
+/// nulls. `NotBetween` then negates valid rows and leaves nulls null —
+/// matching SQL three-valued logic; `sanitize_filter_mask` collapses nulls
+/// into `false` to match the predicate evaluator's "NULL → false" rule.
+fn evaluate_between_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ let (Some(low), Some(high)) = (literals.first(), literals.get(1)) else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(low_scalar) = literal_scalar_for_arrow_filter(low, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(high_scalar) = literal_scalar_for_arrow_filter(high, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let lo_mask = arrow_gt_eq(array, &low_scalar)?;
+ let hi_mask = arrow_lt_eq(array, &high_scalar)?;
+ let between = arrow_arith::boolean::and_kleene(&lo_mask, &hi_mask)?;
+ let result = match op {
+ PredicateOperator::Between => between,
+ PredicateOperator::NotBetween => arrow_arith::boolean::not(&between)?,
+ _ => unreachable!(),
+ };
+ Ok(sanitize_filter_mask(result))
+}
+
+fn evaluate_set_membership_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ if literals.is_empty() {
+ return Ok(match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ });
+ }
+
+ let mut combined = match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ };
+
+ for literal in literals {
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let comparison_op = match op {
+ PredicateOperator::In => PredicateOperator::Eq,
+ PredicateOperator::NotIn => PredicateOperator::NotEq,
+ _ => unreachable!(),
+ };
+ let mask = sanitize_filter_mask(evaluate_column_predicate(array, &scalar, comparison_op)?);
+ combined = combine_filter_masks(&combined, &mask, matches!(op, PredicateOperator::In));
+ }
+
+ Ok(combined)
+}
+
+fn evaluate_column_predicate(
+ column: &ArrayRef,
+ scalar: &Scalar<ArrayRef>,
+ op: PredicateOperator,
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::Eq => arrow_eq(column, scalar),
+ PredicateOperator::NotEq => arrow_neq(column, scalar),
+ PredicateOperator::Lt => arrow_lt(column, scalar),
+ PredicateOperator::LtEq => arrow_lt_eq(column, scalar),
+ PredicateOperator::Gt => arrow_gt(column, scalar),
+ PredicateOperator::GtEq => arrow_gt_eq(column, scalar),
+ PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let pattern = pattern_scalar_for_string_kernel(scalar, column.data_type())?;
+ match op {
+ PredicateOperator::StartsWith => arrow_starts_with(column, &pattern),
+ PredicateOperator::EndsWith => arrow_ends_with(column, &pattern),
+ PredicateOperator::Contains => arrow_contains(column, &pattern),
+ PredicateOperator::Like => arrow_like(column, &pattern),
+ _ => unreachable!(),
+ }
+ }
+ PredicateOperator::IsNull
+ | PredicateOperator::IsNotNull
+ | PredicateOperator::In
+ | PredicateOperator::NotIn
+ | PredicateOperator::Between
+ | PredicateOperator::NotBetween => Ok(BooleanArray::new_null(column.len())),
+ }
+}
+
+/// `arrow_string::like::*` kernels reject mismatched string types — Utf8 column
+/// against Utf8 pattern is fine, but a LargeUtf8 / Utf8View column needs a
+/// pattern of the same flavour. The shared scalar built upstream is always
+/// `StringArray` (Utf8); promote it to match the column when needed.
+fn pattern_scalar_for_string_kernel(
+ scalar: &Scalar<ArrayRef>,
+ column_type: &arrow_schema::DataType,
+) -> Result<Scalar<ArrayRef>, ArrowError> {
+ use arrow_array::{LargeStringArray, StringArray, StringViewArray};
+ use arrow_schema::DataType as ArrowDataType;
+
+ let arr = scalar.get().0;
+ let value = arr
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .and_then(|s| (s.len() == 1 && s.is_valid(0)).then(|| s.value(0).to_string()));
+ let Some(value) = value else {
+ return Ok(scalar.clone());
+ };
+ Ok(match column_type {
+ ArrowDataType::Utf8 => Scalar::new(Arc::new(StringArray::from(vec![value])) as ArrayRef),
+ ArrowDataType::LargeUtf8 => {
+ Scalar::new(Arc::new(LargeStringArray::from(vec![value])) as ArrayRef)
+ }
+ ArrowDataType::Utf8View => {
+ Scalar::new(Arc::new(StringViewArray::from(vec![value])) as ArrayRef)
+ }
+ ArrowDataType::Dictionary(_, value_type) if value_type.as_ref() == &ArrowDataType::Utf8 => {
+ Scalar::new(Arc::new(StringArray::from(vec![value])) as ArrayRef)
+ }
+ other => {
+ return Err(ArrowError::InvalidArgumentError(format!(
+ "string predicate against non-string column type {other:?}"
+ )))
+ }
+ })
+}
+
+/// Collapse `NULL` mask entries to `false`, matching the residual-filter
+/// convention. A mask with no nulls is returned unchanged.
+pub(crate) fn sanitize_filter_mask(mask: BooleanArray) -> BooleanArray {
+ if mask.null_count() == 0 {
+ return mask;
+ }
+
+ boolean_mask_from_predicate(mask.len(), |row_index| {
+ mask.is_valid(row_index) && mask.value(row_index)
+ })
+}
+
+fn combine_filter_masks(left: &BooleanArray, right: &BooleanArray, use_or: bool) -> BooleanArray {
+ debug_assert_eq!(left.len(), right.len());
+ boolean_mask_from_predicate(left.len(), |row_index| {
+ if use_or {
+ left.value(row_index) || right.value(row_index)
+ } else {
+ left.value(row_index) && right.value(row_index)
+ }
+ })
+}
+
+fn boolean_mask_from_predicate(
+ len: usize,
+ mut predicate: impl FnMut(usize) -> bool,
+) -> BooleanArray {
+ BooleanArray::from((0..len).map(&mut predicate).collect::<Vec<_>>())
+}
+
+// ---------------------------------------------------------------------------
+// Literal → Arrow scalar conversion
+// ---------------------------------------------------------------------------
+
+/// Convert a paimon [`Datum`] literal into an Arrow scalar matching
+/// `file_data_type`, or `None` when the literal / type pair is unsupported for
+/// filtering (in which case the caller falls open, keeping the row).
+pub(crate) fn literal_scalar_for_arrow_filter(
+ literal: &Datum,
+ file_data_type: &DataType,
+) -> crate::Result<Option<Scalar<ArrayRef>>> {
+ let array: ArrayRef = match file_data_type {
+ DataType::Boolean(_) => match literal {
+ Datum::Bool(value) => Arc::new(BooleanArray::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::TinyInt(_) => {
+ match integer_literal(literal).and_then(|value| i8::try_from(value).ok()) {
+ Some(value) => Arc::new(Int8Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::SmallInt(_) => {
+ match integer_literal(literal).and_then(|value| i16::try_from(value).ok()) {
+ Some(value) => Arc::new(Int16Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::Int(_) => {
+ match integer_literal(literal).and_then(|value| i32::try_from(value).ok()) {
+ Some(value) => Arc::new(Int32Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::BigInt(_) => {
+ match integer_literal(literal).and_then(|value| i64::try_from(value).ok()) {
+ Some(value) => Arc::new(Int64Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ }
+ }
+ DataType::Float(_) => match float32_literal(literal) {
+ Some(value) => Arc::new(Float32Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ },
+ DataType::Double(_) => match float64_literal(literal) {
+ Some(value) => Arc::new(Float64Array::new_scalar(value).into_inner()),
+ None => return Ok(None),
+ },
+ DataType::Char(_) | DataType::VarChar(_) => match literal {
+ Datum::String(value) => Arc::new(StringArray::new_scalar(value.as_str()).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_) => match literal {
+ Datum::Bytes(value) => Arc::new(BinaryArray::new_scalar(value.as_slice()).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Date(_) => match literal {
+ Datum::Date(value) => Arc::new(Date32Array::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Time(_) => match literal {
+ Datum::Time(value) => Arc::new(Time32MillisecondArray::new_scalar(*value).into_inner()),
+ _ => return Ok(None),
+ },
+ DataType::Timestamp(ts) => match literal {
+ Datum::Timestamp { millis, nanos } => {
+ let Some(array) = timestamp_scalar(*millis, *nanos, ts.precision(), None)? else {
+ return Ok(None);
+ };
+ array
+ }
+ _ => return Ok(None),
+ },
+ DataType::LocalZonedTimestamp(ts) => match literal {
+ Datum::LocalZonedTimestamp { millis, nanos } => {
+ let Some(array) = timestamp_scalar(*millis, *nanos, ts.precision(), Some("UTC"))?
+ else {
+ return Ok(None);
+ };
+ array
+ }
+ _ => return Ok(None),
+ },
+ DataType::Decimal(decimal) => match literal {
Review Comment:
This rejects decimal literals whose scale differs from the file column scale, but Paimon Datum compares decimals by mathematical value across scales. A valid predicate like DECIMAL 1.0 against a DECIMAL(…, 2) column can now fail during residual filtering. Please rescale exactly when possible, or evaluate decimal predicates with the same semantics as Datum.
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1247 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ data_type: predicate_data_type,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ // Evaluate against the predicate's declared type (the table type),
+ // not the file column's type. Under schema evolution these differ
+ // (e.g. an old INT file column for a promoted BIGINT table column):
+ // building the literal in the narrower file type could fail for an
+ // out-of-range value, incorrectly erroring when the exact answer is
+ // simply "no rows". Promotion is always widening, so cast the decoded
+ // column up to the predicate type first — then the literal is always
+ // representable and the comparison is exact.
+ let predicate_arrow_type = crate::arrow::paimon_type_to_arrow(predicate_data_type)?;
+ let mask = if column.data_type() == &predicate_arrow_type {
+ evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ } else {
+ let cast_column = arrow_cast::cast(column, &predicate_arrow_type).map_err(|e| {
+ Error::DataInvalid {
+ message: format!(
+ "Failed to cast residual column '{}' from {:?} to {:?}: {e}",
+ file_field.name(),
+ column.data_type(),
+ predicate_arrow_type
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ evaluate_exact_leaf_predicate(&cast_column, predicate_data_type, *op, literals)
+ }
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let result = evaluate_column_predicate(array, &scalar, op)?;
+ Ok(sanitize_filter_mask(result))
+ }
+ PredicateOperator::Between | PredicateOperator::NotBetween => {
+ evaluate_between_predicate(array, data_type, op, literals)
+ }
+ }
+}
+
+/// `Between` / `NotBetween` translate to `gt_eq(col, low) & lt_eq(col, high)`
+/// (and its negation). `arrow_ord::cmp` produces nullable masks: any null
+/// row makes the comparison null, so a fully-built `Between` mask preserves
+/// nulls. `NotBetween` then negates valid rows and leaves nulls null —
+/// matching SQL three-valued logic; `sanitize_filter_mask` collapses nulls
+/// into `false` to match the predicate evaluator's "NULL → false" rule.
+fn evaluate_between_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ let (Some(low), Some(high)) = (literals.first(), literals.get(1)) else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(low_scalar) = literal_scalar_for_arrow_filter(low, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(high_scalar) = literal_scalar_for_arrow_filter(high, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let lo_mask = arrow_gt_eq(array, &low_scalar)?;
+ let hi_mask = arrow_lt_eq(array, &high_scalar)?;
+ let between = arrow_arith::boolean::and_kleene(&lo_mask, &hi_mask)?;
+ let result = match op {
+ PredicateOperator::Between => between,
+ PredicateOperator::NotBetween => arrow_arith::boolean::not(&between)?,
+ _ => unreachable!(),
+ };
+ Ok(sanitize_filter_mask(result))
+}
+
+fn evaluate_set_membership_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ if literals.is_empty() {
+ return Ok(match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ });
+ }
+
+ let mut combined = match op {
+ PredicateOperator::In => BooleanArray::from(vec![false; array.len()]),
+ PredicateOperator::NotIn => {
+ boolean_mask_from_predicate(array.len(), |row_index| array.is_valid(row_index))
+ }
+ _ => unreachable!(),
+ };
+
+ for literal in literals {
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let comparison_op = match op {
+ PredicateOperator::In => PredicateOperator::Eq,
+ PredicateOperator::NotIn => PredicateOperator::NotEq,
+ _ => unreachable!(),
+ };
+ let mask = sanitize_filter_mask(evaluate_column_predicate(array, &scalar, comparison_op)?);
+ combined = combine_filter_masks(&combined, &mask, matches!(op, PredicateOperator::In));
+ }
+
+ Ok(combined)
+}
+
+fn evaluate_column_predicate(
Review Comment:
This uses Arrow's generic comparison kernels for Binary/VarBinary as well, but Paimon's Datum::Bytes ordering is Java signed-byte order (`0xFF < 0x00`). Arrow compares byte slices unsigned, so binary range predicates can return the wrong rows. Please add a Binary/VarBinary-specific comparison path that matches Paimon's comparator, with a residual test.
##########
crates/paimon/src/table/data_file_reader.rs:
##########
@@ -69,6 +69,26 @@ impl DataFileReader {
self
}
+ /// Reject projecting `_ROW_ID` alongside a data predicate. `_ROW_ID` is
+ /// assigned positionally from post-filter batch row counts, so a residual
+ /// filter that drops rows would desync it. (`_ROW_ID` predicates travel via
+ /// `row_ranges`, not `predicates`, so they do not trip this.)
+ fn reject_row_id_with_predicate(
+ read_type: &[DataField],
+ predicates: &[Predicate],
+ ) -> crate::Result<()> {
+ let projects_row_id = read_type
Review Comment:
This guard treats any predicate as row-dropping. Constant predicates such as AlwaysTrue cannot desync _ROW_ID but are rejected here. Consider normalizing away no-op predicates, or only rejecting predicates that can actually filter rows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "QuakeWang (via GitHub)" <gi...@apache.org>.
QuakeWang commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3522386926
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -223,7 +235,32 @@ impl FormatFileReader for ParquetFormatReader {
}
let batch_stream = batch_stream_builder.build()?;
- Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed())
+
+ if all_enforced {
+ // Fast path: the row filter enforced every predicate exactly during
+ // decode. Return the stream directly — no residual pass.
+ return Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed());
+ }
+
+ // Residual backstop: at least one predicate (`Or`/`Not`/unsupported leaf)
Review Comment:
This residual backstop only works when the compound predicate reaches the format reader. The normal ReadBuilder/TableRead path still passes data predicates through reader_pruning_predicates(), which drops And/Or/Not, so public reads will not actually get exact filtering for these predicates. Please keep the full read predicate for residual filtering and use a separate pruned copy only for stats/pushdown.
##########
crates/paimon/src/arrow/format/mod.rs:
##########
@@ -54,8 +54,13 @@ pub(crate) struct FilePredicates {
/// - Row range selection
#[async_trait]
pub(crate) trait FormatFileReader: Send + Sync {
- /// Read a single data file, returning a stream of RecordBatches
- /// containing only the projected columns (using names from the file's schema).
+ /// Read a single data file, returning a stream of RecordBatches containing
+ /// at least the projected columns (using names from the file's schema). A
+ /// reader MAY include extra columns it needed to scan (e.g. predicate columns
+ /// for residual filtering); the caller (`DataFileReader`) projects to the
+ /// requested output by name, so extra columns are harmless. Each batch MUST
+ /// already have the pushed-down predicate applied exactly — residual filtering
Review Comment:
This contract is currently stronger than the implementations. Blob still ignores predicates, and Mosaic only does row-group stats pruning without row-level residual filtering. Either the contract should be narrowed to the formats changed here, or the remaining readers need to reject/apply predicates exactly.
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1184 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ let mask = evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
Review Comment:
For schema-evolved files this can turn a valid predicate into a read error. Example: table column promoted INT -> BIGINT, old file column is INT, predicate value > i32::MAX. literal_scalar_for_arrow_filter() returns None for the old INT column, but the exact result for that file is simply no rows, not DataInvalid. This needs boundary handling/tests for promoted numeric types.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #448:
URL: https://github.com/apache/paimon-rust/pull/448
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3526265480
##########
crates/paimon/src/table/read_builder.rs:
##########
@@ -233,10 +232,13 @@ impl<'a> ReadBuilder<'a> {
Some(projected) => self.resolve_projected_fields(projected)?,
};
+ // Pass the FULL data predicate through (including `And`/`Or`/`Not`).
+ // Pushdown/stats skip compound nodes; the residual pass enforces the full
+ // predicate exactly. Pruning here would drop compound predicates.
Ok(TableRead::new(
self.table,
read_type,
- reader_pruning_predicates(self.filter.data_predicates.clone()),
+ self.filter.data_predicates.clone(),
Review Comment:
Fixed. `TableRead::read_with_evolution` now passes the data predicates into `DataEvolutionReader`, which filters every batch exactly on both branches (raw-convertible and column-merge): the read set is widened with predicate columns, `_ROW_ID` is attached first, then the residual filter runs and the batch is projected back to the caller's read type. Inner file readers still receive no predicates, so the two branches stay consistent.
Tests go through the real `ReadBuilder` -> `to_arrow` path: leaf and `Or` predicates, cross-file predicate on the merge branch, `_ROW_ID` + predicate (original ids survive), and null-fill semantics (comparisons drop, `IS NULL` keeps).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Apply exact residual filtering for ORC, Avro, and Row reads [paimon-rust]
Posted by "JunRuiLee (via GitHub)" <gi...@apache.org>.
JunRuiLee commented on code in PR #448:
URL: https://github.com/apache/paimon-rust/pull/448#discussion_r3522433673
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -223,7 +235,32 @@ impl FormatFileReader for ParquetFormatReader {
}
let batch_stream = batch_stream_builder.build()?;
- Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed())
+
+ if all_enforced {
+ // Fast path: the row filter enforced every predicate exactly during
+ // decode. Return the stream directly — no residual pass.
+ return Ok(batch_stream.map(|r| r.map_err(Error::from)).boxed());
+ }
+
+ // Residual backstop: at least one predicate (`Or`/`Not`/unsupported leaf)
Review Comment:
Good catch — you're right, and this is the important one. `TableRead::with_filter` runs `reader_pruning_predicates()`, which drops `And`/`Or`/`Not`, so the residual backstop never sees a compound predicate on the public read path (my Parquet test exercised `read_batch_stream` directly and bypassed that). I'll thread two predicate sets through `TableRead`/`DataFileReader`: the full predicate for exact residual filtering, and a separately-pruned copy for stats/row-group pushdown only — and add a test through the real `ReadBuilder`/`TableRead` path (not the format reader directly) so this is actually covered.
##########
crates/paimon/src/arrow/residual.rs:
##########
@@ -0,0 +1,1184 @@
+// 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.
+
+//! Shared, exact Arrow-batch residual predicate evaluator.
+//!
+//! This module holds the single source of truth for evaluating a
+//! [`FilePredicates`] set against an already-decoded Arrow [`RecordBatch`],
+//! producing a boolean row mask (or filtering the batch directly). It replaces
+//! two near-duplicate copies that previously lived in the Parquet and Vortex
+//! format readers:
+//!
+//! - The Parquet copy implemented the full leaf operator set (comparison,
+//! set-membership, and the string / range operators `StartsWith` / `EndsWith`
+//! / `Contains` / `Like` / `Between` / `NotBetween`).
+//! - The Vortex copy carried the compound batch walker (And/Or/Not plus the
+//! `Leaf` arm that maps a `file_field` to the corresponding batch column via
+//! [`same_data_field`]) and a broader literal-to-scalar conversion (Time /
+//! Timestamp / LocalZonedTimestamp / Blob), but deferred the string / range
+//! leaves.
+//!
+//! The consolidation keeps the complete leaf dispatch (from Parquet), the
+//! compound walker + `filter_record_batch_by_predicates` (from Vortex), and the
+//! broader `literal_scalar_for_arrow_filter` (from Vortex). As a result the
+//! string / range leaves are now evaluated everywhere the shared walker runs —
+//! they are no longer silently deferred.
+//!
+//! `NULL` rows collapse to `false` via [`sanitize_filter_mask`], matching the
+//! evaluator's residual-filter convention everywhere.
+//!
+//! The module is always compiled (independent of the `vortex` feature), so it
+//! must not reference any vortex-specific types.
+
+use crate::arrow::format::FilePredicates;
+use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::Error;
+use arrow_array::{
+ Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as ArrowDatum, Decimal128Array,
+ Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array, RecordBatch, Scalar,
+ StringArray, Time32MillisecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+ TimestampNanosecondArray,
+};
+use arrow_ord::cmp::{
+ eq as arrow_eq, gt as arrow_gt, gt_eq as arrow_gt_eq, lt as arrow_lt, lt_eq as arrow_lt_eq,
+ neq as arrow_neq,
+};
+use arrow_schema::ArrowError;
+use arrow_string::like::{
+ contains as arrow_contains, ends_with as arrow_ends_with, like as arrow_like,
+ starts_with as arrow_starts_with,
+};
+use std::sync::Arc;
+
+/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
+///
+/// `scan_fields` describes the columns actually present in `batch` (in order);
+/// `predicates.file_fields` describes the full file schema the predicate indices
+/// point into. Each leaf's `file_field` is resolved to a `batch` column via
+/// [`same_data_field`]. When no predicate produces a mask (e.g. empty predicate
+/// list), the batch is returned unchanged.
+pub(crate) fn filter_record_batch_by_predicates(
+ batch: RecordBatch,
+ predicates: &FilePredicates,
+ scan_fields: &[DataField],
+) -> crate::Result<RecordBatch> {
+ let Some(mask) = evaluate_predicates_mask(
+ &batch,
+ &predicates.predicates,
+ &predicates.file_fields,
+ scan_fields,
+ )?
+ else {
+ return Ok(batch);
+ };
+
+ arrow_select::filter::filter_record_batch(&batch, &mask).map_err(|e| Error::DataInvalid {
+ message: format!("Failed to filter RecordBatch by predicates: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+/// Evaluate the conjunction of `predicates` against `batch`, returning the
+/// combined boolean row mask, or `None` when no predicate contributed a mask
+/// (identity — keep every row).
+pub(crate) fn evaluate_predicates_mask(
+ batch: &RecordBatch,
+ predicates: &[Predicate],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ let mut combined = None;
+ for predicate in predicates {
+ let Some(mask) = evaluate_predicate_mask(batch, predicate, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+}
+
+fn evaluate_predicate_mask(
+ batch: &RecordBatch,
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> crate::Result<Option<BooleanArray>> {
+ match predicate {
+ Predicate::AlwaysTrue => Ok(Some(BooleanArray::from(vec![true; batch.num_rows()]))),
+ Predicate::AlwaysFalse => Ok(Some(BooleanArray::from(vec![false; batch.num_rows()]))),
+ Predicate::And(children) => {
+ let mut combined = None;
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ continue;
+ };
+ combined = Some(match combined {
+ Some(existing) => combine_filter_masks(&existing, &mask, false),
+ None => mask,
+ });
+ }
+ Ok(combined)
+ }
+ Predicate::Or(children) => {
+ let mut combined = BooleanArray::from(vec![false; batch.num_rows()]);
+ for child in children {
+ let Some(mask) = evaluate_predicate_mask(batch, child, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ combined = combine_filter_masks(&combined, &mask, true);
+ }
+ Ok(Some(combined))
+ }
+ Predicate::Not(inner) => {
+ let Some(mask) = evaluate_predicate_mask(batch, inner, file_fields, scan_fields)?
+ else {
+ return Ok(None);
+ };
+ Ok(Some(boolean_mask_from_predicate(mask.len(), |row_index| {
+ !mask.value(row_index)
+ })))
+ }
+ Predicate::Leaf {
+ index,
+ op,
+ literals,
+ ..
+ } => {
+ let Some(file_field) = file_fields.get(*index) else {
+ return Ok(None);
+ };
+ // Resolve the predicate column in the batch by NAME against the batch's
+ // own schema. We must not index by the column's position in
+ // `scan_fields`: a reader's emitted batch may order columns by its file
+ // schema (e.g. ORC `ProjectionMask::named_roots`), not by `scan_fields`
+ // order, so positional indexing can select the wrong column (and
+ // compare mismatched types). `scan_fields` is used only to detect the
+ // Gap-A "predicate column not scanned" bug below.
+ let column = batch
+ .schema()
+ .index_of(file_field.name())
+ .ok()
+ .map(|batch_index| batch.column(batch_index));
+ let Some(column) = column else {
+ // The predicate column exists in the file schema but is absent
+ // from the batch actually scanned — this is the Gap-A bug (a
+ // reader that did not widen its scan to include predicate columns
+ // before filtering). It must never happen. Fail loudly in
+ // debug/test builds; degrade to a skip (rather than panic) in
+ // release. `scan_fields` is unused for resolution now (we look up
+ // by name in the batch), so touch it here only to keep the guard
+ // message informative.
+ let _ = scan_fields;
+ debug_assert!(
+ false,
+ "residual predicate column '{}' exists in file_fields but is missing from the scanned batch; the reader must widen its scan to include predicate columns",
+ file_field.name()
+ );
+ return Ok(None);
+ };
+ let mask = evaluate_exact_leaf_predicate(column, file_field.data_type(), *op, literals)
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Failed to evaluate residual predicate: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(mask))
+ }
+ }
+}
+
+/// Two [`DataField`]s refer to the same logical column when their IDs match, or
+/// (for schemas without stable IDs) their names match.
+pub(crate) fn same_data_field(left: &DataField, right: &DataField) -> bool {
+ left.id() == right.id() || left.name() == right.name()
+}
+
+/// Widen `read_fields` to include every column referenced by `predicates` that
+/// is not already projected, so a reader scans `read_fields ∪ predicate columns`
+/// and the residual filter can see every predicate column.
+///
+/// Deduped by [`same_data_field`]: a predicate column already present in
+/// `read_fields` is not added twice. When `predicates` is `None`, `read_fields`
+/// is returned unchanged.
+pub(crate) fn widen_scan_fields(
+ read_fields: &[DataField],
+ predicates: Option<&FilePredicates>,
+) -> Vec<DataField> {
+ let mut fields = read_fields.to_vec();
+
+ if let Some(fp) = predicates {
+ let mut predicate_indices = Vec::new();
+ for predicate in &fp.predicates {
+ collect_predicate_field_indices(predicate, &mut predicate_indices);
+ }
+ for index in predicate_indices {
+ if let Some(field) = fp.file_fields.get(index) {
+ push_unique_scan_field(&mut fields, field);
+ }
+ }
+ }
+
+ fields
+}
+
+pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices: &mut Vec<usize>) {
+ match predicate {
+ Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ collect_predicate_field_indices(child, indices);
+ }
+ }
+ Predicate::Not(inner) => collect_predicate_field_indices(inner, indices),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
+ }
+}
+
+pub(crate) fn push_unique_scan_field(fields: &mut Vec<DataField>, field: &DataField) {
+ if !fields
+ .iter()
+ .any(|existing| same_data_field(existing, field))
+ {
+ fields.push(field.clone());
+ }
+}
+
+/// Error for a leaf predicate whose literal(s) cannot be converted to an Arrow
+/// scalar for the column's type (e.g. a decimal literal whose scale differs from
+/// the column, or a malformed leaf with too few literals). The residual pass is
+/// the last line of exactness, so it must error rather than pass all rows.
+fn unconvertible_literal_error(op: PredicateOperator, data_type: &DataType) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "residual predicate operator {op:?} has a literal that cannot be evaluated exactly against column type {data_type:?}"
+ ))
+}
+
+/// Evaluate a single leaf predicate against a decoded column, producing an
+/// exact boolean row mask with the `NULL` → `false` convention applied.
+/// This is the *complete* leaf dispatch: comparison, null checks,
+/// set-membership, the string operators (`StartsWith` / `EndsWith` / `Contains`
+/// / `Like`), and the range operators (`Between` / `NotBetween`).
+pub(crate) fn evaluate_exact_leaf_predicate(
+ array: &ArrayRef,
+ data_type: &DataType,
+ op: PredicateOperator,
+ literals: &[Datum],
+) -> Result<BooleanArray, ArrowError> {
+ match op {
+ PredicateOperator::IsNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_null(row_index)
+ })),
+ PredicateOperator::IsNotNull => Ok(boolean_mask_from_predicate(array.len(), |row_index| {
+ array.is_valid(row_index)
+ })),
+ PredicateOperator::In | PredicateOperator::NotIn => {
+ evaluate_set_membership_predicate(array, data_type, op, literals)
+ }
+ PredicateOperator::Eq
+ | PredicateOperator::NotEq
+ | PredicateOperator::Lt
+ | PredicateOperator::LtEq
+ | PredicateOperator::Gt
+ | PredicateOperator::GtEq
+ | PredicateOperator::StartsWith
+ | PredicateOperator::EndsWith
+ | PredicateOperator::Contains
+ | PredicateOperator::Like => {
+ let Some(literal) = literals.first() else {
+ return Err(unconvertible_literal_error(op, data_type));
+ };
+ let Some(scalar) = literal_scalar_for_arrow_filter(literal, data_type)
+ .map_err(|e| ArrowError::ComputeError(e.to_string()))?
+ else {
+ // The literal cannot be converted to an Arrow scalar for this
+ // column type (e.g. a decimal literal whose scale differs from
+ // the column). Erroring is required for correctness: returning
+ // all-true here would silently pass every row (a wrong-read),
+ // and this residual pass is the only place the predicate is
+ // enforced when the row filter did not accept the leaf.
+ return Err(unconvertible_literal_error(op, data_type));
Review Comment:
Agreed — erroring here is wrong for schema-evolved files. For a promoted INT→BIGINT column read from an old INT file, a predicate value outside the old column's range should yield *no rows* for that file, not `DataInvalid`. I'll handle the numeric-range boundary (out-of-range literal → empty result / AlwaysFalse for that file) and only error on genuinely unrepresentable cases, with tests for the promotion boundaries.
##########
crates/paimon/src/arrow/format/mod.rs:
##########
@@ -54,8 +54,13 @@ pub(crate) struct FilePredicates {
/// - Row range selection
#[async_trait]
pub(crate) trait FormatFileReader: Send + Sync {
- /// Read a single data file, returning a stream of RecordBatches
- /// containing only the projected columns (using names from the file's schema).
+ /// Read a single data file, returning a stream of RecordBatches containing
+ /// at least the projected columns (using names from the file's schema). A
+ /// reader MAY include extra columns it needed to scan (e.g. predicate columns
+ /// for residual filtering); the caller (`DataFileReader`) projects to the
+ /// requested output by name, so extra columns are harmless. Each batch MUST
+ /// already have the pushed-down predicate applied exactly — residual filtering
Review Comment:
Right, the contract is stronger than Blob (ignores predicates) and Mosaic (stats pruning only). I'll narrow the doc to the formats changed here (Parquet/ORC/Avro/Row/Vortex) and note that Blob/Mosaic still rely on stats-only pruning, rather than overstating a guarantee they don't provide.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org