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