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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
