This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new d66d3223 fix(predicate): resolve _ROW_ID by name, not by its
placeholder leaf index (#642)
d66d3223 is described below
commit d66d3223159cd72bead6c36cc2531fec8ab35f7b
Author: Jiajia Li <[email protected]>
AuthorDate: Sun Aug 9 21:55:57 2026 +0800
fix(predicate): resolve _ROW_ID by name, not by its placeholder leaf index
(#642)
---
crates/paimon/src/arrow/filtering.rs | 49 ++-
crates/paimon/src/arrow/format/blob.rs | 7 +-
crates/paimon/src/arrow/format/mosaic.rs | 27 ++
crates/paimon/src/arrow/format/orc.rs | 7 +-
crates/paimon/src/arrow/format/parquet.rs | 54 +++-
crates/paimon/src/arrow/format/vortex.rs | 5 +
crates/paimon/src/arrow/residual.rs | 152 ++++++---
crates/paimon/src/predicate_stats.rs | 6 +-
crates/paimon/src/spec/mod.rs | 2 +
crates/paimon/src/spec/predicate.rs | 85 ++++-
crates/paimon/src/spec/schema.rs | 22 ++
crates/paimon/src/table/data_evolution_reader.rs | 237 +++++++++++++-
crates/paimon/src/table/format_table_read.rs | 4 +
crates/paimon/src/table/kv_file_reader.rs | 77 +++++
crates/paimon/src/table/read_builder.rs | 392 +++++++++++++++++++++--
crates/paimon/src/table/row_id_predicate.rs | 182 +++++++----
crates/paimon/src/table/stats_filter.rs | 8 +-
crates/paimon/src/table/table_scan.rs | 60 +++-
crates/paimon/src/table/vector_search_builder.rs | 17 +-
19 files changed, 1213 insertions(+), 180 deletions(-)
diff --git a/crates/paimon/src/arrow/filtering.rs
b/crates/paimon/src/arrow/filtering.rs
index 16170317..b502e7f7 100644
--- a/crates/paimon/src/arrow/filtering.rs
+++ b/crates/paimon/src/arrow/filtering.rs
@@ -17,7 +17,7 @@
use crate::arrow::schema_evolution::create_index_mapping;
pub(crate) use crate::predicate_stats::{predicates_may_match_with_schema,
StatsAccessor};
-use crate::spec::{DataField, Predicate, PredicateOperator};
+use crate::spec::{is_row_id_column, DataField, Predicate, PredicateOperator};
/// Remap predicates from table-level indices to file-level indices.
/// Predicates referencing fields not present in the file are resolved based on
@@ -44,6 +44,12 @@ fn remap_predicate(predicate: &Predicate, mapping:
&[Option<usize>]) -> Predicat
op,
literals,
} => {
+ // `_ROW_ID` is not a file column and has no per-file position, so
+ // mapping its placeholder index would collapse the leaf to a
+ // constant. Keep it; the residual resolves it by name.
+ if is_row_id_column(column) {
+ return predicate.clone();
+ }
match mapping.get(*index).copied().flatten() {
Some(file_index) => Predicate::Leaf {
column: column.clone(),
@@ -134,3 +140,44 @@ fn normalize_field_mapping(mapping: Option<Vec<i32>>,
num_fields: usize) -> Vec<
})
.unwrap_or_else(|| identity_field_mapping(num_fields))
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{DataType, Datum, IntType, PredicateBuilder,
PredicateOperator};
+
+ #[test]
+ fn test_a_row_id_leaf_survives_per_file_remapping() {
+ let table_fields = vec![
+ DataField::new(1, "added".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(0, "base".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let file_fields = vec![table_fields[1].clone()];
+ let leaf = crate::spec::row_id_leaf(PredicateOperator::NotEq,
vec![Datum::Long(102)]);
+
+ assert_eq!(
+ remap_predicates_to_file(std::slice::from_ref(&leaf),
&table_fields, &file_fields),
+ vec![leaf]
+ );
+ }
+
+ #[test]
+ fn test_a_row_id_branch_can_die_during_per_file_remapping() {
+ let table_fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "added".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let file_fields = vec![table_fields[0].clone()];
+ let filter = Predicate::or(vec![
+ PredicateBuilder::new(&table_fields)
+ .is_null("added")
+ .unwrap(),
+ crate::spec::row_id_leaf(PredicateOperator::Eq,
vec![Datum::Long(5)]),
+ ]);
+
+ assert_eq!(
+ remap_predicates_to_file(&[filter], &table_fields, &file_fields),
+ vec![Predicate::AlwaysTrue]
+ );
+ }
+}
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index 080c5056..a7c7d7ed 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -135,10 +135,15 @@ impl FormatFileReader for BlobFormatReader {
reader: Box<dyn FileRead>,
file_size: u64,
read_fields: &[DataField],
- _predicates: Option<&FilePredicates>,
+ predicates: Option<&FilePredicates>,
batch_size: Option<usize>,
row_selection: Option<Vec<RowRange>>,
) -> crate::Result<ArrowRecordBatchStream> {
+ // This reader evaluates no predicate at all, so nothing would enforce
a
+ // `_ROW_ID` one.
+ if let Some(fp) = predicates {
+
crate::table::row_id_predicate::reject_row_id_filter(&fp.predicates, "blob
files")?;
+ }
let field_kind = validate_read_fields(read_fields)?;
let target_schema = build_target_arrow_schema(read_fields)?;
diff --git a/crates/paimon/src/arrow/format/mosaic.rs
b/crates/paimon/src/arrow/format/mosaic.rs
index 356fcc49..304eadc9 100644
--- a/crates/paimon/src/arrow/format/mosaic.rs
+++ b/crates/paimon/src/arrow/format/mosaic.rs
@@ -56,6 +56,11 @@ impl FormatFileReader for MosaicFormatReader {
batch_size: Option<usize>,
row_selection: Option<Vec<RowRange>>,
) -> crate::Result<ArrowRecordBatchStream> {
+ // This reader only prunes row groups by stats, and stats cannot
decide a
+ // `_ROW_ID` predicate, so nothing would enforce it.
+ if let Some(fp) = predicates {
+
crate::table::row_id_predicate::reject_row_id_filter(&fp.predicates, "mosaic
files")?;
+ }
let handle = tokio::runtime::Handle::try_current().map_err(|e|
Error::UnexpectedError {
message: "Mosaic reader requires a Tokio runtime".to_string(),
source: Some(Box::new(e)),
@@ -976,6 +981,28 @@ mod tests {
assert_eq!(ids.value(4), 5);
}
+ #[tokio::test]
+ async fn test_row_id_predicate_is_rejected() {
+ let data = write_mosaic(&sample_batch());
+ let fields = data_fields();
+ let predicates = FilePredicates {
+ predicates: vec![crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::NotEq,
+ vec![Datum::Long(101)],
+ )],
+ row_filter_factory: None,
+ file_fields: fields.clone(),
+ };
+ let err = read_batches_with_predicates(data, &fields,
Some(&predicates), None)
+ .await
+ .unwrap_err();
+
+ assert!(
+ matches!(&err, Error::Unsupported { message } if
message.contains("_ROW_ID")),
+ "unexpected error: {err:?}"
+ );
+ }
+
#[tokio::test]
async fn test_read_projection_order() {
let fields = data_fields();
diff --git a/crates/paimon/src/arrow/format/orc.rs
b/crates/paimon/src/arrow/format/orc.rs
index 8c600b0d..6ebbeb5f 100644
--- a/crates/paimon/src/arrow/format/orc.rs
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -17,7 +17,7 @@
use super::{FilePredicates, FormatFileReader};
use crate::io::FileRead;
-use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::spec::{is_row_id_column, DataField, DataType, Datum, Predicate,
PredicateOperator};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use async_trait::async_trait;
@@ -222,6 +222,7 @@ fn build_orc_leaf_predicate(
file_fields: &[DataField],
) -> Option<orc_rust::predicate::Predicate> {
let Predicate::Leaf {
+ column,
index,
op,
literals,
@@ -230,6 +231,10 @@ fn build_orc_leaf_predicate(
else {
return None;
};
+ // Not in the file, and its index would push the wrong column down.
+ if is_row_id_column(column) {
+ return None;
+ }
let file_field = file_fields.get(*index)?;
let column = file_field.name();
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 58491401..c039074d 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -25,8 +25,8 @@ use crate::arrow::{ParquetReadBudget, RowFilter,
RowFilterContext};
use crate::io::{FileRead, OutputFile};
use crate::spec::stats::BinaryTableStats;
use crate::spec::{
- BinaryRowBuilder, CoreOptions, DataField, DataType, Datum,
MetadataStatsMode, Predicate,
- PredicateOperator,
+ is_row_id_column, BinaryRowBuilder, CoreOptions, DataField, DataType,
Datum, MetadataStatsMode,
+ Predicate, PredicateOperator,
};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
@@ -767,11 +767,11 @@ fn build_parquet_arrow_predicate(
// the union of referenced Parquet roots, ordered exactly as the projected
// RecordBatch. This preserves OR/NOT semantics; splitting it into leaf
// RowFilters would incorrectly turn the expression into a conjunction.
- let mut field_indices = Vec::new();
- crate::arrow::residual::collect_predicate_field_indices(predicate, &mut
field_indices);
- let mut projected = field_indices
+ let mut leaf_refs = Vec::new();
+ crate::arrow::residual::collect_predicate_leaf_refs(predicate, &mut
leaf_refs);
+ let mut projected = leaf_refs
.into_iter()
- .filter_map(|index| {
+ .filter_map(|(_, index)| {
let field = file_fields.get(index)?;
parquet_root_index(parquet_schema, field.name()).map(|root| (root,
field.clone()))
})
@@ -826,11 +826,19 @@ fn parquet_predicate_row_filter_accepted(
match predicate {
Predicate::AlwaysTrue | Predicate::AlwaysFalse => Ok(true),
Predicate::Leaf {
+ column,
index,
op,
literals,
..
- } => parquet_leaf_row_filter_accepted(parquet_schema, *index, *op,
literals, file_fields),
+ } => parquet_leaf_row_filter_accepted(
+ parquet_schema,
+ column,
+ *index,
+ *op,
+ literals,
+ file_fields,
+ ),
Predicate::And(children) | Predicate::Or(children) => {
for child in children {
if !parquet_predicate_row_filter_accepted(parquet_schema,
child, file_fields)? {
@@ -853,6 +861,7 @@ fn parquet_predicate_row_filter_accepted(
/// an unsupported (but well-formed) leaf yields `Ok(false)`.
fn parquet_leaf_row_filter_accepted(
parquet_schema: &parquet::schema::types::SchemaDescriptor,
+ column: &str,
index: usize,
op: PredicateOperator,
literals: &[Datum],
@@ -861,6 +870,11 @@ fn parquet_leaf_row_filter_accepted(
if !predicate_supported_for_parquet_row_filter(op) {
return Ok(false);
}
+ // Not in the file, so the decoder cannot evaluate it. Rejecting the leaf
+ // rejects any enclosing predicate too, leaving it to the post-scan
residual.
+ if is_row_id_column(column) {
+ return Ok(false);
+ }
let Some(file_field) = file_fields.get(index) else {
return Ok(false);
};
@@ -2243,6 +2257,32 @@ mod tests {
assert!(row_filter.is_some());
}
+ #[test]
+ fn test_row_id_predicate_builds_no_decoder_row_filter() {
+ let fields = test_fields();
+ let schema = test_parquet_schema();
+ assert!(build_parquet_row_filter(
+ &schema,
+ &[crate::spec::row_id_leaf(
+ super::PredicateOperator::Eq,
+ vec![Datum::Long(1)]
+ )],
+ &fields
+ )
+ .expect("row filter should build")
+ .is_none());
+
+ let mixed = Predicate::or(vec![
+ crate::spec::row_id_leaf(super::PredicateOperator::Eq,
vec![Datum::Long(1)]),
+ PredicateBuilder::new(&fields)
+ .equal("score", Datum::Int(7))
+ .expect("leaf should build"),
+ ]);
+ assert!(build_parquet_row_filter(&schema, &[mixed], &fields)
+ .expect("row filter should build")
+ .is_none());
+ }
+
// -----------------------------------------------------------------------
// String predicate tests (StartsWith / EndsWith / Contains)
// -----------------------------------------------------------------------
diff --git a/crates/paimon/src/arrow/format/vortex.rs
b/crates/paimon/src/arrow/format/vortex.rs
index 163e510a..7d6d637e 100644
--- a/crates/paimon/src/arrow/format/vortex.rs
+++ b/crates/paimon/src/arrow/format/vortex.rs
@@ -149,6 +149,11 @@ fn read_vortex_batches(
})?;
if scan_fields.is_empty() {
+ // `_ROW_ID` never widens the scan, so this fast path has nothing to
+ // evaluate it against and would treat the predicate as matching.
+ if let Some(fp) = predicates.as_ref() {
+
crate::table::row_id_predicate::reject_row_id_filter(&fp.predicates, "this
read")?;
+ }
let row_count = if constant_predicates_match(predicates.as_ref()) {
match &row_selection {
Some(ranges) => ranges.iter().map(|r| r.count() as
usize).sum(),
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index fe4e9c43..93159af6 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -45,7 +45,7 @@
//! must not reference any vortex-specific types.
use crate::arrow::format::FilePredicates;
-use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::spec::{is_row_id_column, DataField, DataType, Datum, Predicate,
PredicateOperator};
use crate::Error;
use arrow_array::{
Array, ArrayRef, BinaryArray, BooleanArray, Date32Array, Datum as
ArrowDatum, Decimal128Array,
@@ -159,41 +159,51 @@ fn evaluate_predicate_mask(
})))
}
Predicate::Leaf {
+ column,
index,
op,
literals,
data_type: predicate_data_type,
- ..
} => {
- let Some(file_field) = file_fields.get(*index) else {
- return Ok(None);
+ // Resolve the batch column by NAME, but pick that name carefully:
a
+ // real column may be renamed in the file, so its batch name comes
+ // from `file_fields[index]`; `_ROW_ID` is absent from
`file_fields`
+ // and only its own name is meaningful. Java's `PredicateRemapper`
+ // rebinds by name too.
+ let file_field = if is_row_id_column(column) {
+ None
+ } else {
+ match file_fields.get(*index) {
+ Some(field) => Some(field),
+ None => 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
+ let field_name = file_field.map_or(column.as_str(),
DataField::name);
+ let field_type = file_field.map_or(predicate_data_type,
DataField::data_type);
+ // Never index by position in `scan_fields`: a reader's emitted
batch
+ // may order columns by its file schema (e.g. ORC
+ // `ProjectionMask::named_roots`) rather than by `scan_fields`
order.
+ // `scan_fields` is used only for the Gap-A guard below.
+ let array = batch
.schema()
- .index_of(file_field.name())
+ .index_of(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 Some(array) = array else {
let _ = scan_fields;
+ if file_field.is_none() {
+ // Backstop for residuals applied to a predicate-free
reader
+ // (PK merge output, vector search); readers that own their
+ // predicates reject earlier.
+ return
Err(crate::table::row_id_predicate::unsupported_row_id_filter(
+ "this read",
+ ));
+ }
+ // A real column missing here is a reader bug: it did not widen
+ // its scan to the predicate columns.
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()
+ "residual predicate column '{field_name}' is missing from
the scanned batch; the reader must widen its scan to include predicate columns"
);
return Ok(None);
};
@@ -206,16 +216,14 @@ fn evaluate_predicate_mask(
// 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)
+ let mask = if array.data_type() == &predicate_arrow_type {
+ evaluate_exact_leaf_predicate(array, field_type, *op, literals)
} else {
- let cast_column = arrow_cast::cast(column,
&predicate_arrow_type).map_err(|e| {
+ let cast_column = arrow_cast::cast(array,
&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
+ "Failed to cast residual column '{field_name}'
from {:?} to {predicate_arrow_type:?}: {e}",
+ array.data_type()
),
source: Some(Box::new(e)),
}
@@ -251,11 +259,15 @@ pub(crate) fn widen_scan_fields(
let mut fields = read_fields.to_vec();
if let Some(fp) = predicates {
- let mut predicate_indices = Vec::new();
+ let mut refs = Vec::new();
for predicate in &fp.predicates {
- collect_predicate_field_indices(predicate, &mut predicate_indices);
+ collect_predicate_leaf_refs(predicate, &mut refs);
}
- for index in predicate_indices {
+ for (name, index) in refs {
+ // Not read from the file, so there is nothing to widen with.
+ if is_row_id_column(name) {
+ continue;
+ }
if let Some(field) = fp.file_fields.get(index) {
push_unique_scan_field(&mut fields, field);
}
@@ -265,15 +277,22 @@ pub(crate) fn widen_scan_fields(
fields
}
-pub(crate) fn collect_predicate_field_indices(predicate: &Predicate, indices:
&mut Vec<usize>) {
+/// Collect every leaf as `(column name, leaf index)`.
+///
+/// Callers that resolve a leaf positionally must check the name first — see
+/// [`crate::spec::is_row_id_column`].
+pub(crate) fn collect_predicate_leaf_refs<'a>(
+ predicate: &'a Predicate,
+ refs: &mut Vec<(&'a str, usize)>,
+) {
match predicate {
- Predicate::Leaf { index, .. } => indices.push(*index),
+ Predicate::Leaf { column, index, .. } => refs.push((column.as_str(),
*index)),
Predicate::And(children) | Predicate::Or(children) => {
for child in children {
- collect_predicate_field_indices(child, indices);
+ collect_predicate_leaf_refs(child, refs);
}
}
- Predicate::Not(inner) => collect_predicate_field_indices(inner,
indices),
+ Predicate::Not(inner) => collect_predicate_leaf_refs(inner, refs),
Predicate::AlwaysTrue | Predicate::AlwaysFalse => {}
}
}
@@ -972,7 +991,7 @@ fn float64_literal(literal: &Datum) -> Option<f64> {
#[cfg(test)]
mod tests {
use super::*;
- use crate::spec::{IntType, VarCharType};
+ use crate::spec::{row_id_leaf, IntType, VarCharType, ROW_ID_FIELD_NAME};
use arrow_array::{Int32Array, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use std::sync::Arc;
@@ -1454,6 +1473,61 @@ mod tests {
let _ = filter_record_batch_by_predicates(batch, &fp, &scan_fields);
}
+ fn batch_with_row_id(values: Vec<i32>, row_ids: Vec<i64>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("v", ArrowDataType::Int32, true),
+ ArrowField::new(ROW_ID_FIELD_NAME, ArrowDataType::Int64, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(values)),
+ Arc::new(arrow_array::Int64Array::from(row_ids)),
+ ],
+ )
+ .unwrap()
+ }
+
+ #[test]
+ fn test_row_id_leaf_resolves_by_name_not_by_placeholder_index() {
+ let v = int_field(1, "v");
+ let batch = batch_with_row_id(vec![99, 7, 7], vec![1, 2, 3]);
+ let pred = row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(1)]);
+ let fp = file_predicates(vec![pred], vec![v.clone()]);
+ let out = filter_record_batch_by_predicates(batch, &fp, &[v]).unwrap();
+ assert_eq!(int_values(&out), vec![99]);
+ }
+
+ #[test]
+ fn test_row_id_leaf_inside_a_disjunction_resolves_by_name() {
+ let v = int_field(1, "v");
+ let batch = batch_with_row_id(vec![99, 7, 5], vec![1, 2, 3]);
+ let pred = Predicate::or(vec![
+ row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(1)]),
+ leaf(
+ 0,
+ DataType::Int(IntType::new()),
+ PredicateOperator::Eq,
+ vec![Datum::Int(7)],
+ ),
+ ]);
+ let fp = file_predicates(vec![pred], vec![v.clone()]);
+ let out = filter_record_batch_by_predicates(batch, &fp, &[v]).unwrap();
+ assert_eq!(int_values(&out), vec![99, 7]);
+ }
+
+ #[test]
+ fn test_widen_scan_fields_skips_system_columns() {
+ let other = int_field(1, "other");
+ let v = int_field(2, "v");
+ let fp = file_predicates(
+ vec![row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(1)])],
+ vec![other, v.clone()],
+ );
+ let widened = widen_scan_fields(std::slice::from_ref(&v), Some(&fp));
+ assert_eq!(widened, vec![v]);
+ }
+
#[test]
fn test_filter_when_batch_column_order_differs_from_scan_fields() {
// Regression: a reader (e.g. ORC `ProjectionMask::named_roots`) may
emit
diff --git a/crates/paimon/src/predicate_stats.rs
b/crates/paimon/src/predicate_stats.rs
index 71c930dc..57d01e76 100644
--- a/crates/paimon/src/predicate_stats.rs
+++ b/crates/paimon/src/predicate_stats.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::spec::{DataField, DataType, Datum, Predicate, PredicateOperator};
+use crate::spec::{is_row_id_column, DataField, DataType, Datum, Predicate,
PredicateOperator};
use std::cmp::Ordering;
pub(crate) trait StatsAccessor {
@@ -426,6 +426,8 @@ fn predicate_may_match_with_schema<T: StatsAccessor>(
Predicate::Not(inner) => {
!predicate_must_match_with_schema(inner, stats, field_mapping,
file_fields)
}
+ // `_ROW_ID` has no column stats, so never prune on it.
+ Predicate::Leaf { column, .. } if is_row_id_column(column) => true,
Predicate::Leaf {
index,
data_type,
@@ -469,6 +471,8 @@ fn predicate_must_match_with_schema<T: StatsAccessor>(
Predicate::Not(inner) => {
!predicate_may_match_with_schema(inner, stats, field_mapping,
file_fields)
}
+ // Stats cannot decide `_ROW_ID`, so it never provably matches.
+ Predicate::Leaf { column, .. } if is_row_id_column(column) => false,
Predicate::Leaf {
index,
data_type,
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 22780b91..3fb3802e 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -101,6 +101,8 @@ pub(crate) use predicate::datum_cmp;
pub(crate) use predicate::eval_row;
pub(crate) use predicate::extract_datum;
pub(crate) use predicate::like_match;
+#[cfg(test)]
+pub(crate) use predicate::row_id_leaf;
pub use predicate::{
field_idx_to_partition_idx, Datum, Predicate, PredicateBuilder,
PredicateOperator, Transform,
TransformInput,
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index a6d609a6..66946abd 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -26,7 +26,7 @@
use crate::error::*;
use crate::spec::binary_row::BinaryRow;
use crate::spec::types::DataType;
-use crate::spec::DataField;
+use crate::spec::{is_row_id_column, DataField};
use std::cmp::Ordering;
use std::fmt;
@@ -434,6 +434,9 @@ impl Predicate {
op,
literals,
} => {
+ if is_row_id_column(column) {
+ return None;
+ }
let new_index = (*mapping.get(*index)?)?;
Some(Predicate::Leaf {
column: column.clone(),
@@ -472,7 +475,9 @@ impl Predicate {
/// retained as a residual data predicate after partition projection.
pub(crate) fn references_only_mapped_fields(&self, mapping:
&[Option<usize>]) -> bool {
match self {
- Predicate::Leaf { index, .. } =>
mapping.get(*index).is_some_and(Option::is_some),
+ Predicate::Leaf { column, index, .. } => {
+ !is_row_id_column(column) &&
mapping.get(*index).is_some_and(Option::is_some)
+ }
Predicate::And(children) | Predicate::Or(children) => children
.iter()
.all(|child| child.references_only_mapped_fields(mapping)),
@@ -483,6 +488,8 @@ impl Predicate {
/// Project leaf field indices from table schema space into a smaller
field space.
///
+ /// A `_ROW_ID` leaf never maps — see [`is_row_id_column`].
+ ///
/// Unlike [`Self::remap_field_index`], mixed `AND` subtrees keep the
children
/// that can be projected and drop the rest. `OR` and `NOT` still require
all
/// children to be projectable to preserve correctness.
@@ -501,6 +508,9 @@ impl Predicate {
op,
literals,
} => {
+ if is_row_id_column(column) {
+ return None;
+ }
let new_index = (*mapping.get(*index)?)?;
Some(Predicate::Leaf {
column: column.clone(),
@@ -1344,6 +1354,19 @@ fn validate_datum_matches_type(datum: &Datum, data_type:
&DataType) -> Result<()
Ok(())
}
+/// Build a `_ROW_ID` leaf the way callers have to: the column is in no
schema, so
+/// [`PredicateBuilder`] cannot resolve it and the index is a placeholder.
+#[cfg(test)]
+pub(crate) fn row_id_leaf(op: PredicateOperator, literals: Vec<Datum>) ->
Predicate {
+ Predicate::Leaf {
+ column: crate::spec::ROW_ID_FIELD_NAME.to_string(),
+ index: 0,
+ data_type: DataType::BigInt(crate::spec::BigIntType::new()),
+ op,
+ literals,
+ }
+}
+
// ---------------------------------------------------------------------------
// field_idx_to_partition_idx
// ---------------------------------------------------------------------------
@@ -1720,6 +1743,64 @@ mod tests {
]
}
+ #[test]
+ fn test_system_column_never_maps_onto_a_field_space() {
+ let fields = test_fields();
+ let partition_first = [
+ fields[2].clone(),
+ fields[0].clone(),
+ fields[1].clone(),
+ fields[3].clone(),
+ ];
+ let mapping =
+ field_idx_to_partition_idx(&partition_first, &["dt".to_string(),
"hr".to_string()]);
+ assert_eq!(mapping[0], Some(0), "field 0 is the dt partition key");
+
+ let leaf = row_id_leaf(PredicateOperator::GtEq, vec![Datum::Long(10)]);
+ assert_eq!(leaf.project_field_index_inclusive(&mapping), None);
+ assert_eq!(leaf.remap_field_index(&mapping), None);
+ assert!(!leaf.references_only_mapped_fields(&mapping));
+ }
+
+ #[test]
+ fn test_system_column_does_not_drag_a_conjunction_onto_a_field_space() {
+ let fields = test_fields();
+ let mapping = field_idx_to_partition_idx(&fields, &["hr".to_string()]);
+ let dt = PredicateBuilder::new(&fields)
+ .greater_or_equal("hr", Datum::Int(3))
+ .unwrap();
+
+ let conjunction = Predicate::and(vec![
+ row_id_leaf(PredicateOperator::GtEq, vec![Datum::Long(10)]),
+ dt.clone(),
+ ]);
+ let projected =
conjunction.project_field_index_inclusive(&mapping).unwrap();
+ assert!(!matches!(projected, Predicate::And(_)), "only hr survives");
+ assert!(!conjunction.references_only_mapped_fields(&mapping));
+
+ let disjunction = Predicate::or(vec![
+ row_id_leaf(PredicateOperator::GtEq, vec![Datum::Long(10)]),
+ dt,
+ ]);
+ assert_eq!(disjunction.project_field_index_inclusive(&mapping), None);
+ }
+
+ #[test]
+ fn test_only_row_id_is_unresolvable_by_index() {
+ let fields = vec![
+ DataField::new(0, "rowkind".to_string(),
DataType::Int(IntType::new())),
+ DataField::new(1, "hr".to_string(), DataType::Int(IntType::new())),
+ ];
+ let mapping = field_idx_to_partition_idx(&fields,
&["rowkind".to_string()]);
+ let leaf = PredicateBuilder::new(&fields)
+ .equal("rowkind", Datum::Int(1))
+ .unwrap();
+
+ assert!(leaf.project_field_index_inclusive(&mapping).is_some());
+ assert!(leaf.remap_field_index(&mapping).is_some());
+ assert!(leaf.references_only_mapped_fields(&mapping));
+ }
+
// ======================== PredicateBuilder basics
========================
#[test]
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 24a7b498..dd45712e 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -1015,6 +1015,28 @@ pub const VALUE_KIND_FIELD_ID: i32 = i32::MAX - 2;
pub const ROW_KIND_FIELD_NAME: &str = "rowkind";
+/// A row's global id. Nullable: a data file lacking `first_row_id` yields
nulls.
+pub(crate) fn row_id_data_field() -> DataField {
+ DataField::new(
+ ROW_ID_FIELD_ID,
+ ROW_ID_FIELD_NAME.to_string(),
+ DataType::BigInt(crate::spec::BigIntType::with_nullable(true)),
+ )
+}
+
+/// `_ROW_ID` is synthesized by the reader and is not a table column, so
+/// `PredicateBuilder` cannot resolve it and callers hand-build the leaf with a
+/// placeholder index. Every index-based resolution must recognize it by name
+/// instead, or it binds the predicate to whatever field sits at that index.
+///
+/// The other reserved names are excluded on purpose. `_SEQUENCE_NUMBER` and
+/// `_VALUE_KIND` are physical columns of a KV file and do have a position;
none
+/// of them is ever referenced by a predicate. Widening this to every reserved
+/// name would instead break a schema that predates their rejection.
+pub(crate) fn is_row_id_column(name: &str) -> bool {
+ name == ROW_ID_FIELD_NAME
+}
+
/// Must match Java Paimon's `SpecialFields.ROW_KIND` (Integer.MAX_VALUE - 4).
pub const ROW_KIND_FIELD_ID: i32 = i32::MAX - 4;
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 2aa6d485..cf04829b 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -27,8 +27,7 @@ use crate::arrow::{build_target_arrow_schema,
ParquetReadBudget};
use crate::deletion_vector::{DeletionVector, DeletionVectorFactory};
use crate::io::FileIO;
use crate::spec::{
- BigIntType, BlobDescriptor, BlobViewStruct, DataField, DataFileMeta,
DataType, Predicate,
- ROW_ID_FIELD_ID, ROW_ID_FIELD_NAME,
+ BlobDescriptor, BlobViewStruct, DataField, DataFileMeta, DataType,
Predicate, ROW_ID_FIELD_NAME,
};
use crate::table::dedicated_format_file_writer::is_blob_file_name;
use crate::table::schema_manager::SchemaManager;
@@ -132,7 +131,7 @@ impl DataEvolutionReader {
blob_view_resolve_enabled: bool,
blob_view_rest_env: Option<RESTEnv>,
) -> crate::Result<Self> {
- let row_id_index = read_type.iter().position(|f| f.name() ==
ROW_ID_FIELD_NAME);
+ let projected_row_id_index = read_type.iter().position(|f| f.name() ==
ROW_ID_FIELD_NAME);
let file_read_type: Vec<DataField> = read_type
.iter()
.filter(|f| f.name() != ROW_ID_FIELD_NAME)
@@ -153,11 +152,25 @@ impl DataEvolutionReader {
let wide_file_read_type =
crate::arrow::residual::widen_scan_fields(&file_read_type,
file_predicates.as_ref());
// Wide batches at the _ROW_ID attach point: original read_type columns
- // (caller order, _ROW_ID at row_id_index) followed by the extras.
- // row_id_index <= file_read_type.len(), so inserting _ROW_ID never
- // displaces a trailing extra column.
+ // (caller order, _ROW_ID at row_id_index) followed by the extras. A
+ // projected row_id_index is <= file_read_type.len(), so inserting
+ // _ROW_ID never displaces a trailing extra column.
let mut wide_read_type = read_type;
wide_read_type.extend_from_slice(&wide_file_read_type[file_read_type.len()..]);
+ // A residual on `_ROW_ID` needs the column when `filter_wide_batch`
+ // runs, even unprojected. `widen_scan_fields` cannot supply a
+ // synthesized column, so append it and let `project_output` trim it.
+ let row_id_index = match projected_row_id_index {
+ Some(index) => Some(index),
+ None if predicates
+ .iter()
+ .any(super::row_id_predicate::references_row_id) =>
+ {
+ wide_read_type.push(crate::spec::row_id_data_field());
+ Some(wide_read_type.len() - 1)
+ }
+ None => None,
+ };
let wide_output_schema = build_target_arrow_schema(&wide_read_type)?;
Ok(Self {
@@ -413,8 +426,9 @@ impl DataEvolutionReader {
///
/// Layout invariant: the first `output_schema.fields().len()` columns of
/// `batch` are exactly the original read_type columns. Extras were
appended
- /// at the end by `widen_scan_fields`, and `_ROW_ID` insertion at
- /// `row_id_index` keeps them trailing.
+ /// at the end — by `widen_scan_fields`, plus an unprojected `_ROW_ID` a
+ /// residual needs — and `_ROW_ID` insertion at `row_id_index` keeps them
+ /// trailing.
fn project_output(&self, filtered: RecordBatch) ->
crate::Result<RecordBatch> {
let final_width = self.output_schema.fields().len();
if filtered.num_columns() == final_width {
@@ -919,6 +933,11 @@ fn predicate_references_any_field(
) -> bool {
match predicate {
Predicate::Leaf { column, index, .. } => {
+ // Never a BLOB column; resolving its placeholder index would force
+ // every BLOB to be resolved before filtering.
+ if crate::spec::is_row_id_column(column) {
+ return false;
+ }
field_names.contains(column)
|| table_fields
.get(*index)
@@ -1083,7 +1102,7 @@ impl BlobViewLookup {
let table = table.copy_with_options(options);
let row_ranges = row_ranges_for_blob_view_refs(&refs);
let mut read_builder = table.new_read_builder();
- read_builder.with_read_type(vec![field.clone(), row_id_field()]);
+ read_builder.with_read_type(vec![field.clone(),
crate::spec::row_id_data_field()]);
read_builder.with_row_ranges(row_ranges);
let plan = read_builder.new_scan().plan().await?;
let read = read_builder.new_read()?;
@@ -1197,14 +1216,6 @@ fn row_ranges_for_blob_view_refs(refs:
&[BlobViewStruct]) -> Vec<RowRange> {
ranges
}
-fn row_id_field() -> DataField {
- DataField::new(
- ROW_ID_FIELD_ID,
- ROW_ID_FIELD_NAME.to_string(),
- DataType::BigInt(BigIntType::with_nullable(true)),
- )
-}
-
#[allow(clippy::too_many_arguments)]
fn open_source_stream(
split: &DataSplit,
@@ -6988,6 +6999,198 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_evolution_read_applies_row_id_residual_to_the_row_ids() {
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+
+ let parquet_path = bucket_dir.join("data.parquet");
+ write_int_parquet_file(&parquet_path, vec![("id", vec![1, 2, 3, 4])],
None);
+
+ let table = two_col_evolution_table(table_path);
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file_meta_with_path(
+ "data.parquet",
+ 100,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ )])
+ .build()
+ .unwrap();
+
+ let predicate = crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::NotEq,
+ vec![Datum::Long(102)],
+ );
+
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ builder.with_filter(predicate);
+ let read = builder.new_read().unwrap();
+ let batches = read
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(collect_int_values(&batches, "id"), vec![1, 2, 4]);
+ assert_eq!(batches[0].num_columns(), 1);
+ }
+
+ #[test]
+ fn test_row_id_filter_is_not_blob_dependent() {
+ let table_fields = vec![
+ DataField::new(0, "payload".to_string(),
DataType::Blob(BlobType::new())),
+ DataField::new(1, "v".to_string(), DataType::Int(IntType::new())),
+ ];
+ let blob_fields: HashSet<String> =
["payload".to_string()].into_iter().collect();
+ let row_id = crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::Between,
+ vec![Datum::Long(10), Datum::Long(20)],
+ );
+
+ assert!(!predicate_references_any_field(
+ &row_id,
+ &blob_fields,
+ &table_fields
+ ));
+ let on_blob = PredicateBuilder::new(&table_fields)
+ .is_null("payload")
+ .unwrap();
+ assert!(predicate_references_any_field(
+ &on_blob,
+ &blob_fields,
+ &table_fields
+ ));
+ }
+
+ #[tokio::test]
+ async fn
test_evolution_read_enforces_a_row_id_leaf_nested_in_a_disjunction() {
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+ let parquet_path = bucket_dir.join("data.parquet");
+ write_int_parquet_file(&parquet_path, vec![("id", vec![1, 2, 3, 4])],
None);
+
+ let table = two_col_evolution_table(table_path);
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file_meta_with_path(
+ "data.parquet",
+ 100,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ )])
+ .build()
+ .unwrap();
+
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ let rid = |op, v| crate::spec::row_id_leaf(op, vec![Datum::Long(v)]);
+ builder.with_filter(Predicate::and(vec![
+ rid(crate::spec::PredicateOperator::GtEq, 100),
+ Predicate::or(vec![
+ Predicate::and(vec![
+ rid(crate::spec::PredicateOperator::Eq, 999),
+ pb.equal("id", Datum::Int(2)).unwrap(),
+ ]),
+ pb.equal("id", Datum::Int(4)).unwrap(),
+ ]),
+ ]));
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(collect_int_values(&batches, "id"), vec![4]);
+ }
+
+ #[tokio::test]
+ async fn test_row_id_filter_without_data_evolution_is_rejected() {
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+ let parquet_path = bucket_dir.join("data.parquet");
+ write_int_parquet_file(&parquet_path, vec![("id", vec![1, 2, 3, 4])],
None);
+
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .build()
+ .unwrap(),
+ );
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "plain_t"),
+ table_path,
+ table_schema,
+ None,
+ );
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file_meta_with_path(
+ "data.parquet",
+ 100,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ )])
+ .build()
+ .unwrap();
+
+ let predicate = crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::NotEq,
+ vec![Datum::Long(102)],
+ );
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ builder.with_filter(predicate);
+ let err = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap_err();
+
+ assert!(
+ matches!(&err, crate::Error::Unsupported { message } if
message.contains("_ROW_ID")),
+ "unexpected error: {err:?}"
+ );
+ }
+
/// _ROW_ID + predicate, merge branch: same guarantee across a column
merge.
#[tokio::test]
async fn test_evolution_read_row_id_with_predicate_merge_branch() {
diff --git a/crates/paimon/src/table/format_table_read.rs
b/crates/paimon/src/table/format_table_read.rs
index 9adb2e83..9f0814d6 100644
--- a/crates/paimon/src/table/format_table_read.rs
+++ b/crates/paimon/src/table/format_table_read.rs
@@ -104,6 +104,10 @@ impl<'a> FormatTableRead<'a> {
) -> crate::Result<ArrowRecordBatchStream> {
let core_options = self.table.schema().core_options();
core_options.ensure_read_authorized()?;
+ // Mapping the conjunct onto the data fields drops it, so the read
would
+ // silently ignore the filter. Guard on the read path, not the builder:
+ // `TableRead` is public and can be constructed and filtered directly.
+ super::row_id_predicate::reject_row_id_filter(&self.data_predicates,
"format tables")?;
let read_type = self.read_type.clone();
let output_schema = build_target_arrow_schema(&read_type)?;
let partition_keys = self.table.schema().partition_keys().to_vec();
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index a9612d2a..ee0c3ecc 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -315,6 +315,12 @@ impl KeyValueFileReader {
}
pub fn read(self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
+ // A projected `_ROW_ID` is synthesized as all-nulls here, so the
residual
+ // would silently drop every row rather than hit its missing-column
guard.
+ super::row_id_predicate::reject_row_id_filter(
+ &self.config.predicates,
+ "primary-key tables",
+ )?;
// Build the internal read type for thin-mode files.
// Physical file schema: [_SEQUENCE_NUMBER, _VALUE_KIND,
all_user_cols...]
// We need: _SEQ + _VK + union(read_type, primary_keys)
@@ -703,6 +709,77 @@ mod tests {
use parquet::file::properties::WriterProperties;
use std::sync::Arc;
+ #[tokio::test]
+ async fn test_row_id_filter_on_a_primary_key_table_is_rejected() {
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_row_id_filter";
+ setup_dirs(&file_io, table_path).await;
+ let table = pk_table(&file_io, table_path, &[]);
+
+ write_commit(
+ &table,
+ &int_batch(vec![1, 2, 3], vec![Some(10), Some(20), Some(30)]),
+ )
+ .await;
+ write_commit(
+ &table,
+ &int_batch(vec![1, 2, 3], vec![Some(11), Some(21), Some(31)]),
+ )
+ .await;
+
+ let row_id = crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::NotEq,
+ vec![Datum::Long(102)],
+ );
+ let mut read_builder = table.new_read_builder();
+ read_builder
+ .with_projection(&["id", "value", crate::spec::ROW_ID_FIELD_NAME])
+ .unwrap();
+ read_builder.with_filter(row_id);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let err = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .err()
+ .expect("a _ROW_ID filter must be rejected");
+
+ assert!(
+ matches!(&err, Error::Unsupported { message } if
message.contains("_ROW_ID")),
+ "unexpected error: {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_row_id_conjunct_is_not_treated_as_a_primary_key_conjunct() {
+ let fields = vec![
+ DataField::new(0, "k".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "v".to_string(), DataType::Int(IntType::new())),
+ ];
+ let row_id =
+ crate::spec::row_id_leaf(crate::spec::PredicateOperator::Eq,
vec![Datum::Long(5)]);
+ let on_key = PredicateBuilder::new(&fields)
+ .equal("k", Datum::Int(1))
+ .unwrap();
+
+ assert_eq!(
+ retain_primary_key_conjuncts(
+ std::slice::from_ref(&row_id),
+ &fields,
+ &["k".to_string()]
+ ),
+ Vec::new()
+ );
+ assert_eq!(
+ retain_primary_key_conjuncts(
+ &[Predicate::and(vec![row_id, on_key.clone()])],
+ &fields,
+ &["k".to_string()],
+ ),
+ vec![on_key]
+ );
+ }
+
fn test_file_io() -> FileIO {
FileIOBuilder::new("memory").build().unwrap()
}
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index cce6838e..4af47790 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -308,7 +308,11 @@ struct PaimonReadBuilder<'a> {
projection_names: Option<Vec<String>>,
filter: NormalizedFilter,
limit: Option<usize>,
- row_ranges: Option<Vec<RowRange>>,
+ /// Ranges the caller set. Each `with_row_ranges` replaces them.
+ explicit_row_ranges: Option<Vec<RowRange>>,
+ /// Ranges the current filter yields, recomputed whenever it is replaced.
+ /// Kept apart so neither setter discards the other's constraint.
+ derived_row_ranges: Option<Vec<RowRange>>,
case_sensitive: bool,
parquet_read_budget: Option<Arc<crate::arrow::ParquetReadBudget>>,
}
@@ -321,7 +325,8 @@ impl<'a> PaimonReadBuilder<'a> {
projection_names: None,
filter: NormalizedFilter::default(),
limit: None,
- row_ranges: None,
+ explicit_row_ranges: None,
+ derived_row_ranges: None,
case_sensitive: true,
parquet_read_budget: None,
}
@@ -384,6 +389,7 @@ impl<'a> PaimonReadBuilder<'a> {
/// the full predicate with an exact post-merge residual filter.
pub fn with_filter(&mut self, filter: Predicate) -> &mut Self {
self.filter = normalize_filter(self.table, filter);
+ self.derived_row_ranges = None;
self.try_extract_row_id_ranges();
self
}
@@ -402,25 +408,43 @@ impl<'a> PaimonReadBuilder<'a> {
/// Set row ID ranges `[from, to]` (inclusive). An empty vector selects no
rows.
pub fn with_row_ranges(&mut self, ranges: Vec<RowRange>) -> &mut Self {
- self.row_ranges = Some(ranges);
+ self.explicit_row_ranges = Some(super::merge_row_ranges(ranges));
self
}
- /// Extract `_ROW_ID` predicates from data_predicates into row_ranges.
- /// Only runs when no explicit row_ranges have been set.
+ /// Both are independent restrictions, so a scan honours their
intersection.
+ /// `None` on a side means unconstrained by it.
+ fn effective_row_ranges(&self) -> Option<Vec<RowRange>> {
+ match (&self.explicit_row_ranges, &self.derived_row_ranges) {
+ (Some(explicit), Some(derived)) => Some(
+ super::row_id_predicate::intersect_sorted_ranges(explicit,
derived),
+ ),
+ (Some(only), None) | (None, Some(only)) => Some(only.clone()),
+ (None, None) => None,
+ }
+ }
+
+ /// Derive row ranges from the `_ROW_ID` conjuncts. Independent of any
+ /// explicit ranges — the two compose.
fn try_extract_row_id_ranges(&mut self) {
- if self.row_ranges.is_some() || self.filter.data_predicates.is_empty()
{
+ if self.filter.data_predicates.is_empty() {
+ return;
+ }
+ // Ranges select by `first_row_id`, which only row tracking assigns.
+ // Without it nothing enforces a derived range, so a conjunct handed
over
+ // and dropped from the residual would return every row.
+ let core_options = CoreOptions::new(self.table.schema().options());
+ if !core_options.row_tracking_enabled() &&
!core_options.data_evolution_enabled() {
return;
}
let combined = Predicate::and(self.filter.data_predicates.clone());
if let Some(ranges) =
super::row_id_predicate::extract_row_id_ranges(&combined) {
- self.row_ranges = Some(ranges);
- self.filter.data_predicates = self
- .filter
+ self.derived_row_ranges = Some(ranges);
+ // Ranges are a superset, so a conjunct leaves the residual only
when
+ // they represent it exactly.
+ self.filter
.data_predicates
- .iter()
- .filter_map(super::row_id_predicate::remove_row_id_filter)
- .collect();
+ .retain(|conjunct|
!super::row_id_predicate::ranges_represent_conjunct(conjunct));
}
}
@@ -466,7 +490,7 @@ impl<'a> PaimonReadBuilder<'a> {
self.filter.data_predicates.clone(),
self.filter.bucket_predicate.clone(),
self.limit,
- self.row_ranges.clone(),
+ self.effective_row_ranges(),
)
.with_projected_read_field_ids(projected_read_field_ids_with_predicates(
&read_type,
@@ -606,11 +630,7 @@ pub(super) fn resolve_projected_fields(
}
if name == crate::spec::ROW_ID_FIELD_NAME {
- resolved.push(DataField::new(
- crate::spec::ROW_ID_FIELD_ID,
- crate::spec::ROW_ID_FIELD_NAME.to_string(),
-
crate::spec::DataType::BigInt(crate::spec::BigIntType::with_nullable(true)),
- ));
+ resolved.push(crate::spec::row_id_data_field());
continue;
}
@@ -659,11 +679,16 @@ fn projected_read_field_ids_with_predicates(
table_fields: &[DataField],
) -> Option<HashSet<i32>> {
let mut field_ids = projected_read_field_ids(read_type)?;
- let mut predicate_indices = Vec::new();
+ let mut refs = Vec::new();
for predicate in predicates {
- crate::arrow::residual::collect_predicate_field_indices(predicate,
&mut predicate_indices);
+ crate::arrow::residual::collect_predicate_leaf_refs(predicate, &mut
refs);
}
- for index in predicate_indices {
+ for (name, index) in refs {
+ // Not in `table_fields`, and excluded from this set anyway — files
never
+ // list it in `write_cols`.
+ if crate::spec::is_row_id_column(name) {
+ continue;
+ }
let Some(field) = table_fields.get(index) else {
// A malformed predicate must not make scan planning discard files.
return None;
@@ -757,6 +782,27 @@ mod tests {
)
}
+ fn row_id_table(row_tracking: bool, data_evolution: bool) -> Table {
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let mut builder = Schema::builder()
+ .column("dt", DataType::VarChar(VarCharType::string_type()))
+ .column("id", DataType::Int(IntType::new()))
+ .partition_keys(["dt"]);
+ if row_tracking {
+ builder = builder.option("row-tracking.enabled", "true");
+ }
+ if data_evolution {
+ builder = builder.option("data-evolution.enabled", "true");
+ }
+ Table::new(
+ file_io,
+ Identifier::new("default", "t"),
+ "/tmp/test-read-builder-rowid".to_string(),
+ TableSchema::new(0, &builder.build().unwrap()),
+ None,
+ )
+ }
+
fn simple_table() -> Table {
let file_io = FileIOBuilder::new("file").build().unwrap();
let table_schema = TableSchema::new(
@@ -783,7 +829,10 @@ mod tests {
let mut builder = table.new_read_builder();
builder.with_row_ranges(Vec::new());
- assert_eq!(paimon_builder(&builder).row_ranges, Some(Vec::new()));
+ assert_eq!(
+ paimon_builder(&builder).effective_row_ranges(),
+ Some(Vec::new())
+ );
}
#[tokio::test]
@@ -953,6 +1002,298 @@ mod tests {
);
}
+ use crate::table::RowRange;
+
+ fn row_id_leaf(op: crate::spec::PredicateOperator, v: i64) -> Predicate {
+ crate::spec::row_id_leaf(op, vec![crate::spec::Datum::Long(v)])
+ }
+
+ #[test]
+ fn test_replacing_a_filter_drops_the_range_it_derived() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let mut builder = PaimonReadBuilder::new(&table);
+
+ builder.with_filter(row_id_leaf(PredicateOperator::Eq, 10));
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(10, 10)])
+ );
+
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", crate::spec::Datum::Int(7))
+ .unwrap(),
+ );
+ assert_eq!(builder.effective_row_ranges(), None);
+ }
+
+ #[test]
+ fn test_range_setters_do_not_consume_each_other() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let mut builder = PaimonReadBuilder::new(&table);
+
+ builder.with_filter(row_id_leaf(PredicateOperator::GtEq, 10));
+ builder.with_row_ranges(vec![RowRange::new(0, 20)]);
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(10, 20)])
+ );
+
+ builder.with_row_ranges(vec![RowRange::new(0, 30)]);
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(10, 30)])
+ );
+
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", crate::spec::Datum::Int(7))
+ .unwrap(),
+ );
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(0, 30)])
+ );
+ }
+
+ #[test]
+ fn test_an_explicit_range_survives_a_filter_replacement() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let mut builder = PaimonReadBuilder::new(&table);
+
+ builder.with_row_ranges(vec![RowRange::new(0, 20)]);
+ builder.with_filter(row_id_leaf(PredicateOperator::Eq, 10));
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(10, 10)])
+ );
+
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", crate::spec::Datum::Int(7))
+ .unwrap(),
+ );
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(0, 20)])
+ );
+ }
+
+ #[test]
+ fn test_inexactly_extracted_row_id_conjuncts_stay_residuals() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let mut builder = PaimonReadBuilder::new(&table);
+
+ let contradiction = Predicate::and(vec![
+ row_id_leaf(PredicateOperator::Eq, 1),
+ row_id_leaf(PredicateOperator::NotEq, 1),
+ ]);
+ builder.with_filter(contradiction);
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(1, 1)])
+ );
+ assert_eq!(
+ builder.filter.data_predicates,
+ vec![row_id_leaf(PredicateOperator::NotEq, 1)]
+ );
+
+ let mut exact = PaimonReadBuilder::new(&table);
+ exact.with_filter(row_id_leaf(PredicateOperator::GtEq, 10));
+ assert_eq!(
+ exact.effective_row_ranges(),
+ Some(vec![RowRange::new(10, i64::MAX)])
+ );
+ assert!(exact.filter.data_predicates.is_empty());
+ }
+
+ #[test]
+ fn test_a_mixed_conjunct_is_never_replaced_by_ranges() {
+ use crate::spec::{Datum, PredicateOperator};
+ let table = row_id_table(true, false);
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let disjunction = Predicate::or(vec![
+ Predicate::and(vec![
+ row_id_leaf(PredicateOperator::Eq, 1),
+ pb.equal("id", Datum::Int(7)).unwrap(),
+ ]),
+ pb.equal("dt", Datum::String("x".into())).unwrap(),
+ ]);
+ let mut builder = PaimonReadBuilder::new(&table);
+ builder.with_filter(Predicate::and(vec![
+ row_id_leaf(PredicateOperator::GtEq, 10),
+ disjunction.clone(),
+ ]));
+
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(10, i64::MAX)])
+ );
+ assert_eq!(builder.filter.data_predicates, vec![disjunction]);
+ }
+
+ #[test]
+ fn test_a_nested_empty_row_id_branch_does_not_widen_the_ranges() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let unsatisfiable = Predicate::or(vec![
+ row_id_leaf(PredicateOperator::Gt, i64::MAX),
+ row_id_leaf(PredicateOperator::Lt, 0),
+ ]);
+ let mut builder = PaimonReadBuilder::new(&table);
+ builder.with_filter(Predicate::or(vec![
+ Predicate::and(vec![row_id_leaf(PredicateOperator::Eq, 6),
unsatisfiable]),
+ row_id_leaf(PredicateOperator::Eq, 7),
+ ]));
+
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(7, 7)])
+ );
+ assert!(builder.filter.data_predicates.is_empty());
+ }
+
+ #[test]
+ fn test_a_format_table_rejects_a_row_id_filter() {
+ use crate::spec::PredicateOperator;
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("type", "format-table")
+ .option("file.format", "parquet")
+ .build()
+ .unwrap(),
+ );
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "ft"),
+ "memory:/ft".to_string(),
+ table_schema,
+ None,
+ );
+ let mut builder = ReadBuilder::new(&table);
+ builder.with_filter(row_id_leaf(PredicateOperator::NotEq, 5));
+ let err = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(&[])
+ .err()
+ .expect("a _ROW_ID filter must be rejected");
+
+ assert!(matches!(err, crate::Error::Unsupported { .. }));
+ }
+
+ #[test]
+ fn test_an_empty_row_id_disjunction_stays_a_residual() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let unsatisfiable = Predicate::or(vec![
+ row_id_leaf(PredicateOperator::Gt, i64::MAX),
+ row_id_leaf(PredicateOperator::Lt, 0),
+ ]);
+ let mut builder = PaimonReadBuilder::new(&table);
+ builder.with_filter(Predicate::and(vec![
+ row_id_leaf(PredicateOperator::Eq, 5),
+ unsatisfiable.clone(),
+ ]));
+
+ assert_eq!(builder.effective_row_ranges(), Some(Vec::new()));
+ assert_eq!(builder.filter.data_predicates, vec![unsatisfiable]);
+ }
+
+ #[test]
+ fn test_extraction_needs_a_table_whose_files_carry_row_ids() {
+ use crate::spec::PredicateOperator;
+ let plain_table = row_id_table(false, false);
+ let mut plain = PaimonReadBuilder::new(&plain_table);
+ plain.with_filter(row_id_leaf(PredicateOperator::GtEq, 102));
+ assert_eq!(
+ plain.effective_row_ranges(),
+ None,
+ "nothing to derive ranges from"
+ );
+ assert_eq!(
+ plain.filter.data_predicates.len(),
+ 1,
+ "the conjunct must stay so the read can reject it"
+ );
+
+ for (row_tracking, data_evolution) in [(true, false), (false, true),
(true, true)] {
+ let table = row_id_table(row_tracking, data_evolution);
+ let mut builder = PaimonReadBuilder::new(&table);
+ builder.with_filter(row_id_leaf(PredicateOperator::GtEq, 102));
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(102, i64::MAX)]),
+ "row_tracking={row_tracking} data_evolution={data_evolution}"
+ );
+ assert!(builder.filter.data_predicates.is_empty());
+ }
+ }
+
+ #[test]
+ fn test_an_explicit_range_intersects_a_derived_one() {
+ use crate::spec::PredicateOperator;
+ let table = row_id_table(true, false);
+ let mut builder = PaimonReadBuilder::new(&table);
+ builder.with_filter(Predicate::or(vec![
+ row_id_leaf(PredicateOperator::Eq, 1),
+ row_id_leaf(PredicateOperator::Eq, 8),
+ ]));
+ assert!(
+ builder.filter.data_predicates.is_empty(),
+ "handed to ranges"
+ );
+
+ builder.with_row_ranges(vec![RowRange::new(0, 5)]);
+
+ assert_eq!(
+ builder.effective_row_ranges(),
+ Some(vec![RowRange::new(1, 1)]),
+ "row 8 is outside the explicit range, row 1 satisfies both"
+ );
+ }
+
+ #[test]
+ fn test_row_id_filter_never_projects_onto_a_partition_key() {
+ let table = simple_table();
+ let row_id = crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::GtEq,
+ vec![crate::spec::Datum::Long(10)],
+ );
+
+ let (partition_predicate, data_predicates) =
+ super::split_scan_predicates(&table, row_id.clone());
+ assert_eq!(partition_predicate, None);
+ assert_eq!(data_predicates, vec![row_id.clone()]);
+
+ assert!(!ReadBuilder::new(&table).is_exact_filter_pushdown(&row_id));
+ }
+
+ #[test]
+ fn test_projected_read_field_ids_ignore_system_predicate_fields() {
+ let fields = vec![
+ DataField::new(1, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(2, "payload".to_string(),
DataType::Int(IntType::new())),
+ ];
+ let read_type = Some(vec![fields[1].clone()]);
+ let row_id = crate::spec::row_id_leaf(
+ crate::spec::PredicateOperator::Eq,
+ vec![crate::spec::Datum::Long(1)],
+ );
+
+ assert_eq!(
+ super::projected_read_field_ids_with_predicates(&read_type,
&[row_id], &fields),
+ Some(HashSet::from([2]))
+ );
+ }
+
#[test]
fn test_with_projection_validates_unknown_projection() {
// A column that cannot match under any case sensitivity is an obvious
@@ -1635,6 +1976,7 @@ mod tests {
&Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
.build()
.unwrap(),
);
@@ -1670,8 +2012,8 @@ mod tests {
// _ROW_ID predicates should be extracted into row_ranges
let inner = paimon_builder(&builder);
- assert!(inner.row_ranges.is_some());
- let ranges = inner.row_ranges.as_ref().unwrap();
+ assert!(inner.effective_row_ranges().is_some());
+ let ranges = inner.effective_row_ranges().unwrap();
assert_eq!(ranges.len(), 1);
assert_eq!(ranges[0].from(), 10);
assert_eq!(ranges[0].to(), 20);
@@ -1716,7 +2058,7 @@ mod tests {
builder.with_filter(filter);
// Explicit row_ranges should be preserved, not overwritten
- let ranges = paimon_builder(&builder).row_ranges.as_ref().unwrap();
+ let ranges = paimon_builder(&builder).effective_row_ranges().unwrap();
assert_eq!(ranges.len(), 1);
assert_eq!(ranges[0].from(), 0);
assert_eq!(ranges[0].to(), 5);
diff --git a/crates/paimon/src/table/row_id_predicate.rs
b/crates/paimon/src/table/row_id_predicate.rs
index 353ed69c..b6253fef 100644
--- a/crates/paimon/src/table/row_id_predicate.rs
+++ b/crates/paimon/src/table/row_id_predicate.rs
@@ -19,7 +19,7 @@
//!
//! Reference:
[org.apache.paimon.predicate.RowIdPredicateVisitor](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/predicate/RowIdPredicateVisitor.java)
-use crate::spec::{Datum, Predicate, PredicateOperator, ROW_ID_FIELD_NAME};
+use crate::spec::{is_row_id_column, Datum, Predicate, PredicateOperator};
use crate::table::RowRange;
/// Extract row ranges from `_ROW_ID` predicates in the given filter.
@@ -31,7 +31,7 @@ pub(crate) fn extract_row_id_ranges(predicate: &Predicate) ->
Option<Vec<RowRang
op,
literals,
..
- } if column == ROW_ID_FIELD_NAME => leaf_to_ranges(*op, literals),
+ } if is_row_id_column(column) => leaf_to_ranges(*op, literals),
Predicate::And(children) => {
// AND: intersect all _ROW_ID ranges
let mut result: Option<Vec<RowRange>> = None;
@@ -46,48 +46,82 @@ pub(crate) fn extract_row_id_ranges(predicate: &Predicate)
-> Option<Vec<RowRang
result
}
Predicate::Or(children) => {
- // OR: union all _ROW_ID ranges (all children must have _ROW_ID
predicates)
+ // OR: union all _ROW_ID ranges (all children must have _ROW_ID
predicates).
+ // An empty union is `Some(vec![])`, not `None`: the branches
convert
+ // to "no rows", and reporting that as "nothing extracted" would
let
+ // an enclosing AND skip the subtree and overstate its ranges.
let mut all_ranges: Vec<RowRange> = Vec::new();
for child in children {
let ranges = extract_row_id_ranges(child)?;
all_ranges.extend(ranges);
}
- if all_ranges.is_empty() {
- None
- } else {
- Some(super::merge_row_ranges(all_ranges))
- }
+ Some(super::merge_row_ranges(all_ranges))
}
_ => None,
}
}
-/// Remove `_ROW_ID` predicates from a filter, returning the remaining filter.
-/// Returns `None` if the entire filter is a `_ROW_ID` predicate.
-pub(crate) fn remove_row_id_filter(predicate: &Predicate) -> Option<Predicate>
{
+/// Whether the extracted row ranges represent `conjunct` exactly, so it can be
+/// dropped from the residual.
+///
+/// Two things must hold. The conjunct must be built entirely from convertible
+/// `_ROW_ID` conditions — extraction ignores whatever it cannot convert, so
its
+/// ranges are merely a superset and anything mixed leaves a remainder they do
+/// not carry. And extraction must actually have produced ranges *for this
+/// conjunct*: an `Or` whose branches are all empty yields `None`, which the
+/// enclosing `And` silently skips, so its ranges say nothing about it.
+pub(crate) fn ranges_represent_conjunct(conjunct: &Predicate) -> bool {
+ convertible_row_id_only(conjunct)
+ && extract_row_id_ranges(conjunct).is_some_and(|ranges|
!ranges.is_empty())
+}
+
+fn convertible_row_id_only(predicate: &Predicate) -> bool {
match predicate {
- Predicate::Leaf { column, .. } if column == ROW_ID_FIELD_NAME => None,
- Predicate::And(children) => {
- let filtered: Vec<Predicate> =
- children.iter().filter_map(remove_row_id_filter).collect();
- match filtered.len() {
- 0 => None,
- 1 => Some(filtered.into_iter().next().unwrap()),
- _ => Some(Predicate::and(filtered)),
- }
+ Predicate::Leaf {
+ column,
+ op,
+ literals,
+ ..
+ } => is_row_id_column(column) && leaf_to_ranges(*op,
literals).is_some(),
+ Predicate::And(children) | Predicate::Or(children) => {
+ children.iter().all(convertible_row_id_only)
}
- Predicate::Or(children) => {
- let filtered: Vec<Predicate> =
- children.iter().filter_map(remove_row_id_filter).collect();
- if filtered.len() != children.len() {
- // If any child was entirely _ROW_ID, the OR semantics change;
- // conservatively keep the whole OR.
- Some(predicate.clone())
- } else {
- Some(Predicate::or(filtered))
- }
+ _ => false,
+ }
+}
+
+/// The error for a read that cannot evaluate a `_ROW_ID` predicate. `read`
names
+/// the kind of read.
+///
+/// Only data-evolution reads attach the column; everywhere else the predicate
is
+/// unenforceable, and every alternative is a silent wrong answer — a dropped
+/// conjunct, an all-null synthesized column, or a scan skipped entirely.
+pub(crate) fn unsupported_row_id_filter(read: &str) -> crate::Error {
+ crate::Error::Unsupported {
+ message: format!(
+ "filtering on '_ROW_ID' is not supported by {read}; it is
available on \
+ data-evolution reads, or via row ranges"
+ ),
+ }
+}
+
+/// Reject a `_ROW_ID` predicate on a read that cannot synthesize row ids.
+pub(crate) fn reject_row_id_filter(predicates: &[Predicate], read: &str) ->
crate::Result<()> {
+ if predicates.iter().any(references_row_id) {
+ return Err(unsupported_row_id_filter(read));
+ }
+ Ok(())
+}
+
+/// Whether any leaf of `predicate` references `_ROW_ID`.
+pub(crate) fn references_row_id(predicate: &Predicate) -> bool {
+ match predicate {
+ Predicate::Leaf { column, .. } => is_row_id_column(column),
+ Predicate::And(children) | Predicate::Or(children) => {
+ children.iter().any(references_row_id)
}
- other => Some(other.clone()),
+ Predicate::Not(inner) => references_row_id(inner),
+ Predicate::AlwaysTrue | Predicate::AlwaysFalse => false,
}
}
@@ -131,10 +165,12 @@ fn leaf_to_ranges(op: PredicateOperator, literals:
&[Datum]) -> Option<Vec<RowRa
Some(vec![RowRange::new(0, v - 1)])
}
PredicateOperator::In => {
+ // EVERY literal, or none: ranges standing for only part of the
leaf
+ // would still read as the whole of it and drop the conjunct.
let mut ranges: Vec<RowRange> = literals
.iter()
- .filter_map(|d| datum_to_i64(d).map(|v| RowRange::new(v, v)))
- .collect();
+ .map(|d| datum_to_i64(d).map(|v| RowRange::new(v, v)))
+ .collect::<Option<Vec<_>>>()?;
if ranges.is_empty() {
return None;
}
@@ -146,6 +182,11 @@ fn leaf_to_ranges(op: PredicateOperator, literals:
&[Datum]) -> Option<Vec<RowRa
}
/// Intersect two sorted range lists.
+/// Intersect two range lists that are already sorted and merged.
+pub(crate) fn intersect_sorted_ranges(a: &[RowRange], b: &[RowRange]) ->
Vec<RowRange> {
+ intersect_range_lists(a, b)
+}
+
fn intersect_range_lists(a: &[RowRange], b: &[RowRange]) -> Vec<RowRange> {
let mut result = Vec::new();
let (mut i, mut j) = (0, 0);
@@ -166,17 +207,42 @@ fn intersect_range_lists(a: &[RowRange], b: &[RowRange])
-> Vec<RowRange> {
#[cfg(test)]
mod tests {
+
+ #[test]
+ fn test_a_partially_convertible_in_is_not_converted_at_all() {
+ let partial = row_id_leaf(
+ PredicateOperator::In,
+ vec![Datum::Long(5), Datum::String("7".into())],
+ );
+ assert_eq!(extract_row_id_ranges(&partial), None);
+ assert!(!ranges_represent_conjunct(&partial));
+
+ let whole = row_id_leaf(PredicateOperator::In, vec![Datum::Long(5),
Datum::Long(7)]);
+ assert_eq!(
+ extract_row_id_ranges(&whole),
+ Some(vec![RowRange::new(5, 5), RowRange::new(7, 7)])
+ );
+ assert!(ranges_represent_conjunct(&whole));
+ }
use super::*;
+ use crate::spec::row_id_leaf;
use crate::spec::{BigIntType, DataType};
- fn row_id_leaf(op: PredicateOperator, literals: Vec<Datum>) -> Predicate {
- Predicate::Leaf {
- column: ROW_ID_FIELD_NAME.to_string(),
+ #[test]
+ fn test_references_row_id_finds_nested_leaves() {
+ let other = Predicate::Leaf {
+ column: "v".to_string(),
index: 0,
data_type: DataType::BigInt(BigIntType::new()),
- op,
- literals,
- }
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::Long(7)],
+ };
+ let nested = Predicate::Not(Box::new(Predicate::or(vec![
+ row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(1)]),
+ other.clone(),
+ ])));
+ assert!(references_row_id(&nested));
+ assert!(!references_row_id(&other));
}
fn data_leaf() -> Predicate {
@@ -240,24 +306,26 @@ mod tests {
}
#[test]
- fn test_remove_row_id_filter_leaf() {
- let p = row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(10)]);
- assert!(remove_row_id_filter(&p).is_none());
- }
-
- #[test]
- fn test_remove_row_id_filter_and() {
- let p = Predicate::and(vec![
- row_id_leaf(PredicateOperator::GtEq, vec![Datum::Long(10)]),
+ fn test_ranges_represent_only_a_pure_row_id_conjunct() {
+ assert!(ranges_represent_conjunct(&row_id_leaf(
+ PredicateOperator::GtEq,
+ vec![Datum::Long(10)]
+ )));
+ assert!(!ranges_represent_conjunct(&row_id_leaf(
+ PredicateOperator::NotEq,
+ vec![Datum::Long(10)]
+ )));
+ assert!(!ranges_represent_conjunct(&data_leaf()));
+ assert!(!ranges_represent_conjunct(&Predicate::or(vec![
+ Predicate::and(vec![
+ row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(1)]),
+ data_leaf(),
+ ]),
data_leaf(),
- ]);
- let result = remove_row_id_filter(&p).unwrap();
- assert_eq!(result, data_leaf());
- }
-
- #[test]
- fn test_remove_row_id_filter_keeps_non_row_id() {
- let p = data_leaf();
- assert_eq!(remove_row_id_filter(&p).unwrap(), data_leaf());
+ ])));
+ assert!(ranges_represent_conjunct(&Predicate::or(vec![
+ row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(1)]),
+ row_id_leaf(PredicateOperator::Eq, vec![Datum::Long(2)]),
+ ])));
}
}
diff --git a/crates/paimon/src/table/stats_filter.rs
b/crates/paimon/src/table/stats_filter.rs
index 2eb18e55..7de875c5 100644
--- a/crates/paimon/src/table/stats_filter.rs
+++ b/crates/paimon/src/table/stats_filter.rs
@@ -23,7 +23,9 @@ use crate::predicate_stats::{
data_leaf_may_match, data_leaf_must_match, missing_field_may_match,
missing_field_must_match,
predicates_may_match_with_schema, StatsAccessor,
};
-use crate::spec::{extract_datum, BinaryRow, DataField, DataFileMeta, DataType,
Datum, Predicate};
+use crate::spec::{
+ extract_datum, is_row_id_column, BinaryRow, DataField, DataFileMeta,
DataType, Datum, Predicate,
+};
use std::collections::HashMap;
use std::sync::Arc;
@@ -399,6 +401,8 @@ fn data_evolution_predicate_may_match(
file_stats,
row_count,
),
+ // `_ROW_ID` has no column stats, so never prune a group on it.
+ Predicate::Leaf { column, .. } if is_row_id_column(column) => true,
Predicate::Leaf {
index,
data_type,
@@ -462,6 +466,8 @@ fn data_evolution_predicate_must_match(
file_stats,
row_count,
),
+ // Stats cannot decide `_ROW_ID`, so it never provably matches.
+ Predicate::Leaf { column, .. } if is_row_id_column(column) => false,
Predicate::Leaf {
index,
data_type,
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 43922255..471606a1 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -384,13 +384,9 @@ fn split_row_ranges_for_files(
.flat_map(|file| intersect_ranges_with_file(ranges, file))
.collect(),
);
- if split_ranges.is_empty() {
- return Err(crate::Error::DataInvalid {
- message: "Planned data-evolution split does not overlap selected
row ranges"
- .to_string(),
- source: None,
- });
- }
+ // Empty means this split holds none of the selected rows. Only data
+ // evolution prunes such splits away up front (`RowRangeIndex`), so on a
+ // row-tracking table they reach here legitimately and the caller drops
them.
Ok(Some(split_ranges))
}
@@ -1462,10 +1458,16 @@ impl<'a> PaimonTableScan<'a> {
.to_string(),
});
}
- if self.row_ranges.is_some() {
+ // Both forms: without `first_row_id` a filter stays an ordinary data
+ // predicate instead of becoming a row range.
+ if self.row_ranges.is_some()
+ || self
+ .data_predicates
+ .iter()
+ .any(super::row_id_predicate::references_row_id)
+ {
return Err(crate::Error::Unsupported {
- message: "Batch incremental Diff does not support _ROW_ID
row-range filters"
- .to_string(),
+ message: "Batch incremental Diff does not support _ROW_ID
filters".to_string(),
});
}
// A limit hint cannot be pushed into either side of a Diff: truncating
@@ -1966,6 +1968,12 @@ impl<'a> PaimonTableScan<'a> {
let split_row_ranges =
split_row_ranges_for_files(effective_row_ranges.as_deref(), &file_group)?;
+ if split_row_ranges
+ .as_deref()
+ .is_some_and(<[RowRange]>::is_empty)
+ {
+ continue;
+ }
let mut builder = DataSplitBuilder::new()
.with_snapshot(snapshot_id)
@@ -2272,9 +2280,11 @@ mod tests {
if message.contains("missing or invalid row-id range")));
let outside = make_evo_file("outside", 1, 2, 0, Some(10));
- let error = split_row_ranges_for_files(Some(&ranges),
&[outside]).unwrap_err();
- assert!(matches!(error, Error::DataInvalid { message, .. }
- if message.contains("does not overlap selected row ranges")));
+ assert_eq!(
+ split_row_ranges_for_files(Some(&ranges), &[outside]).unwrap(),
+ Some(Vec::new()),
+ "a split holding none of the selected rows is dropped, not an
error"
+ );
}
fn data_evolution_test_table(table_path: &str, schema: TableSchema) ->
Table {
@@ -3788,6 +3798,30 @@ mod tests {
));
}
+ #[test]
+ fn test_data_evolution_group_is_not_pruned_by_a_row_id_predicate() {
+ let fields = int_field();
+ let file = test_data_file_meta(
+ int_stats_row(Some(10)),
+ int_stats_row(Some(20)),
+ vec![Some(0)],
+ 5,
+ );
+ let row_id =
+ crate::spec::row_id_leaf(crate::spec::PredicateOperator::GtEq,
vec![Datum::Long(100)]);
+
+ assert!(data_evolution_group_matches_predicates(
+ std::slice::from_ref(&file),
+ std::slice::from_ref(&row_id),
+ &fields,
+ ));
+ assert!(data_evolution_group_matches_predicates(
+ &[file],
+ &[Predicate::negate(row_id)],
+ &fields,
+ ));
+ }
+
#[test]
fn test_data_evolution_group_matches_not_prunes_when_inner_must_match() {
let fields = int_field();
diff --git a/crates/paimon/src/table/vector_search_builder.rs
b/crates/paimon/src/table/vector_search_builder.rs
index 3866c6a8..1295db93 100644
--- a/crates/paimon/src/table/vector_search_builder.rs
+++ b/crates/paimon/src/table/vector_search_builder.rs
@@ -23,8 +23,8 @@ use crate::lumina::{
is_lumina_index_type, LuminaIndexMeta, LuminaVectorIndexOptions,
LuminaVectorMetric,
};
use crate::spec::{
- BigIntType, CoreOptions, DataField, DataType, FileKind,
GlobalIndexSearchMode, IndexFileMeta,
- IndexManifest, IndexManifestEntry, Predicate, ROW_ID_FIELD_ID,
ROW_ID_FIELD_NAME,
+ row_id_data_field, CoreOptions, DataField, DataType, FileKind,
GlobalIndexSearchMode,
+ IndexFileMeta, IndexManifest, IndexManifestEntry, Predicate,
ROW_ID_FIELD_NAME,
};
use crate::table::bucket_filter::split_partition_and_data_predicates;
use crate::table::data_file_reader::DataFileReader;
@@ -2148,19 +2148,6 @@ pub(crate) fn reorder_and_strip_position(
Ok(vec![projected])
}
-/// The `_ROW_ID` field to append to a data-evolution read type so the reader
-/// fills each row's global id. Mirrors the field the `DataEvolutionReader`
-/// recognizes (Int64 / `BigInt`, nullable): a data file lacking `first_row_id`
-/// yields nulls here, which `attach_scores_by_row_id` then fails loud on
rather
-/// than mis-aligning scores.
-fn row_id_data_field() -> DataField {
- DataField::new(
- ROW_ID_FIELD_ID,
- ROW_ID_FIELD_NAME.to_string(),
- DataType::BigInt(BigIntType::with_nullable(true)),
- )
-}
-
/// Collect materialized DE rows, join each row's `(rank, score)` by its global
/// `_ROW_ID`, reorder to the search rank order, append the
`__paimon_search_score`
/// column, and drop `_ROW_ID`. Every row must map to a search candidate and
the
