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