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 464904a3 Support nested ROW predicates in native table reads (#1153)
464904a3 is described below
commit 464904a3fda4138fbc9c79b400824cfad0060d8f
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 12:54:42 2026 +0800
Support nested ROW predicates in native table reads (#1153)
---
bindings/python/src/predicate.rs | 53 +--
crates/paimon/src/arrow/filtering.rs | 17 +
crates/paimon/src/arrow/format/orc.rs | 1 +
crates/paimon/src/arrow/format/parquet.rs | 2 +
crates/paimon/src/arrow/format/vortex.rs | 2 +-
crates/paimon/src/arrow/residual.rs | 40 +-
crates/paimon/src/arrow/residual/nested.rs | 494 ++++++++++++++++++++
crates/paimon/src/file_index/evaluator.rs | 3 +
.../paimon/src/file_index/file_index_predicate.rs | 1 +
crates/paimon/src/predicate_stats.rs | 2 +
crates/paimon/src/spec/mod.rs | 4 +-
crates/paimon/src/spec/predicate.rs | 496 +++++++++++++++++----
crates/paimon/src/spec/predicate/field_ref.rs | 106 +++++
.../paimon/src/spec/predicate/field_ref_tests.rs | 380 ++++++++++++++++
crates/paimon/src/table/data_evolution_reader.rs | 228 ++++++++--
crates/paimon/src/table/data_file_reader.rs | 214 ++++++++-
.../src/table/full_text_search_builder/filter.rs | 3 +
crates/paimon/src/table/global_index_scanner.rs | 1 +
.../src/table/global_index_scanner/evaluator.rs | 4 +
crates/paimon/src/table/managed_blob_reader.rs | 105 ++++-
crates/paimon/src/table/partition_filter.rs | 1 +
crates/paimon/src/table/query_auth.rs | 4 +
crates/paimon/src/table/query_auth/rules.rs | 7 +
crates/paimon/src/table/row_id_predicate.rs | 1 +
crates/paimon/src/table/stats_filter.rs | 2 +
crates/paimon/src/table/table_read.rs | 378 +++++++++++++++-
crates/paimon/src/table/table_update_delete.rs | 1 +
27 files changed, 2341 insertions(+), 209 deletions(-)
diff --git a/bindings/python/src/predicate.rs b/bindings/python/src/predicate.rs
index 85d066e5..cb7726f7 100644
--- a/bindings/python/src/predicate.rs
+++ b/bindings/python/src/predicate.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::error::to_py_err;
use paimon::spec::{
DataField, DataType, Datum, DecimalType, Predicate, PredicateBuilder,
TableSchema,
};
@@ -422,6 +423,8 @@ const METHOD_NOT_SUPPORTED: &[&str] = &["not"];
/// The dict shape mirrors the Python predicate tree:
/// - Leaf: `{"method": <op>, "field": <name>, "literals": [..]}`
/// - Compound: `{"method": "and"|"or", "children": [<dict>, ..]}`
+/// - Nested ROW leaf: the same leaf dict with `field` naming the root and
+/// `nested_path` containing separate descendant names (not a dotted string).
///
/// Field types are resolved authoritatively from `fields` (the table schema);
any
/// `index`/`data_type` present in the dict is ignored. Literal conversion is
@@ -488,37 +491,6 @@ pub(crate) fn dict_to_table_predicate(
dict_to_predicate(node, &schema.fields_with_read_metadata(),
case_sensitive)
}
-/// Resolve a leaf's field name to its schema [`DataType`] under the given case
-/// sensitivity.
-///
-/// Case-sensitive: exact match. Case-insensitive: ASCII-fold and require a
-/// unique match. Absent → `ValueError("Column '{field}' not found in
schema")`;
-/// ambiguous (2+ fields collide under folding) → `ValueError` naming the
clash.
-fn resolve_leaf_field_type(
- fields: &[DataField],
- field: &str,
- case_sensitive: bool,
-) -> PyResult<DataType> {
- let not_found = || PyValueError::new_err(format!("Column '{field}' not
found in schema"));
- if case_sensitive {
- return fields
- .iter()
- .find(|f| f.name() == field)
- .map(|f| f.data_type().clone())
- .ok_or_else(not_found);
- }
- let mut matches = fields
- .iter()
- .filter(|f| f.name().eq_ignore_ascii_case(field));
- let first = matches.next().ok_or_else(not_found)?;
- if matches.next().is_some() {
- return Err(PyValueError::new_err(format!(
- "Ambiguous column '{field}': multiple schema fields match
case-insensitively"
- )));
- }
- Ok(first.data_type().clone())
-}
-
/// Convert a single leaf dict (already known not to be `and`/`or`) into a
/// [`Predicate`], resolving the field type from the schema.
fn leaf_to_predicate(
@@ -527,16 +499,21 @@ fn leaf_to_predicate(
fields: &[DataField],
case_sensitive: bool,
) -> PyResult<Predicate> {
- let field: String = node
+ let field_name: String = node
.get_item("field")?
.ok_or_else(|| PyValueError::new_err(format!("'{method}' leaf requires
'field'")))?
.extract()?;
-
- // Resolve field DataType from schema (authoritative) for literal
conversion.
- let data_type = resolve_leaf_field_type(fields, &field, case_sensitive)?;
-
- let literals_obj = node.get_item("literals")?;
let pb = PredicateBuilder::new_with_case_sensitive(fields, case_sensitive);
+ let field = if let Some(path) = node.get_item("nested_path")? {
+ let path = path.extract::<Vec<String>>()?;
+ let names = path.iter().map(String::as_str).collect::<Vec<_>>();
+ pb.nested_field(&field_name, &names)
+ } else {
+ pb.field(&field_name)
+ }
+ .map_err(to_py_err)?;
+ let data_type = field.data_type().clone();
+ let literals_obj = node.get_item("literals")?;
// Convert literals (DataType-driven), wrapping NotImplemented type
messages
// with field context.
@@ -553,7 +530,7 @@ fn leaf_to_predicate(
}
out.push(Some(
predicate_datum(&item, &data_type)
- .map_err(|e| with_field_context(e, &field,
&data_type))?,
+ .map_err(|e| with_field_context(e, &field_name,
&data_type))?,
));
}
}
diff --git a/crates/paimon/src/arrow/filtering.rs
b/crates/paimon/src/arrow/filtering.rs
index 3ae911a2..7fd2ced0 100644
--- a/crates/paimon/src/arrow/filtering.rs
+++ b/crates/paimon/src/arrow/filtering.rs
@@ -37,6 +37,23 @@ pub(crate) fn remap_predicates_to_file(
fn remap_predicate(predicate: &Predicate, mapping: &[Option<usize>]) ->
Predicate {
match predicate {
+ Predicate::NestedLeaf {
+ root,
+ index,
+ path,
+ op,
+ literals,
+ } => match mapping.get(*index).copied().flatten() {
+ Some(index) => Predicate::NestedLeaf {
+ root: root.clone(),
+ index,
+ path: path.clone(),
+ op: *op,
+ literals: literals.clone(),
+ },
+ None if *op == PredicateOperator::IsNull => Predicate::AlwaysTrue,
+ None => Predicate::AlwaysFalse,
+ },
Predicate::Leaf {
column,
index,
diff --git a/crates/paimon/src/arrow/format/orc.rs
b/crates/paimon/src/arrow/format/orc.rs
index 79bd02f9..86115163 100644
--- a/crates/paimon/src/arrow/format/orc.rs
+++ b/crates/paimon/src/arrow/format/orc.rs
@@ -477,6 +477,7 @@ fn build_orc_predicate_inner(
mode: CompoundPredicateMode,
) -> Option<orc_rust::predicate::Predicate> {
match predicate {
+ Predicate::NestedLeaf { .. } => None,
Predicate::Leaf { .. } => build_orc_leaf_predicate(predicate,
file_fields),
Predicate::And(children) => build_orc_and_predicate(children,
file_fields, mode),
Predicate::Or(children) => build_orc_or_predicate(children,
file_fields),
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 33ddbfae..aa2df217 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -1753,6 +1753,7 @@ fn parquet_predicate_row_filter_accepted(
file_fields: &[DataField],
) -> crate::Result<bool> {
match predicate {
+ Predicate::NestedLeaf { .. } => Ok(false),
Predicate::AlwaysTrue | Predicate::AlwaysFalse => Ok(true),
Predicate::Leaf {
column,
@@ -2601,6 +2602,7 @@ fn build_predicate_page_selection(
/// through compound nodes). Indices are already file-level (see caller).
fn collect_leaf_field_indices(predicate: &Predicate, out: &mut Vec<usize>) {
match predicate {
+ Predicate::NestedLeaf { index, .. } => out.push(*index),
Predicate::Leaf { index, .. } => out.push(*index),
Predicate::And(children) | Predicate::Or(children) => {
for child in children {
diff --git a/crates/paimon/src/arrow/format/vortex.rs
b/crates/paimon/src/arrow/format/vortex.rs
index b53437f3..2075b3ab 100644
--- a/crates/paimon/src/arrow/format/vortex.rs
+++ b/crates/paimon/src/arrow/format/vortex.rs
@@ -273,7 +273,7 @@ fn constant_predicate_value(predicate: &Predicate) ->
Option<bool> {
(!saw_unknown).then_some(false)
}
Predicate::Not(inner) => constant_predicate_value(inner).map(|value|
!value),
- Predicate::Leaf { .. } => None,
+ Predicate::Leaf { .. } | Predicate::NestedLeaf { .. } => None,
}
}
diff --git a/crates/paimon/src/arrow/residual.rs
b/crates/paimon/src/arrow/residual.rs
index d4a083e6..71211a86 100644
--- a/crates/paimon/src/arrow/residual.rs
+++ b/crates/paimon/src/arrow/residual.rs
@@ -67,6 +67,8 @@ use arrow_string::like::{
};
use std::sync::Arc;
+mod nested;
+
/// Filter a [`RecordBatch`] to exactly the rows satisfying `predicates`.
///
/// `scan_fields` describes the columns actually present in `batch` (in order);
@@ -118,13 +120,20 @@ pub(crate) fn evaluate_predicates_mask(
Ok(combined)
}
-fn evaluate_predicate_mask(
+pub(crate) fn evaluate_predicate_mask(
batch: &RecordBatch,
predicate: &Predicate,
file_fields: &[DataField],
scan_fields: &[DataField],
) -> crate::Result<Option<BooleanArray>> {
match predicate {
+ Predicate::NestedLeaf {
+ root,
+ path,
+ op,
+ literals,
+ ..
+ } => nested::evaluate(batch, root, path, *op, literals, file_fields,
scan_fields).map(Some),
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) => {
@@ -284,11 +293,39 @@ pub(crate) fn widen_scan_fields(
push_unique_scan_field(&mut fields, field);
}
}
+ for predicate in &fp.predicates {
+ widen_nested_fields(predicate, &fp.file_fields, &mut fields);
+ }
}
fields
}
+fn widen_nested_fields(
+ predicate: &Predicate,
+ file_fields: &[DataField],
+ fields: &mut Vec<DataField>,
+) {
+ match predicate {
+ Predicate::NestedLeaf { root, .. } => {
+ if let Some(source) = file_fields.iter().find(|f| f.id() ==
root.id()) {
+ if let Some(index) = fields.iter().position(|f| f.id() ==
root.id()) {
+ fields[index] = source.clone();
+ } else {
+ fields.push(source.clone());
+ }
+ }
+ }
+ Predicate::And(children) | Predicate::Or(children) => {
+ for child in children {
+ widen_nested_fields(child, file_fields, fields);
+ }
+ }
+ Predicate::Not(child) => widen_nested_fields(child, file_fields,
fields),
+ _ => {}
+ }
+}
+
/// Collect every leaf as `(column name, leaf index)`.
///
/// Callers that resolve a leaf positionally must check the name first — see
@@ -298,6 +335,7 @@ pub(crate) fn collect_predicate_leaf_refs<'a>(
refs: &mut Vec<(&'a str, usize)>,
) {
match predicate {
+ Predicate::NestedLeaf { root, index, .. } => refs.push((root.name(),
*index)),
Predicate::Leaf { column, index, .. } => refs.push((column.as_str(),
*index)),
Predicate::And(children) | Predicate::Or(children) => {
for child in children {
diff --git a/crates/paimon/src/arrow/residual/nested.rs
b/crates/paimon/src/arrow/residual/nested.rs
new file mode 100644
index 00000000..e12f735d
--- /dev/null
+++ b/crates/paimon/src/arrow/residual/nested.rs
@@ -0,0 +1,494 @@
+// 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.
+
+//! Exact nested ROW predicates. Metadata pruning must never interpret the
+//! enclosing ROW's statistics as statistics of a descendant.
+
+use crate::spec::{DataField, DataType, Datum, PredicateOperator};
+use crate::{Error, Result};
+use arrow_array::{
+ make_array, new_null_array, Array, ArrayRef, BooleanArray, RecordBatch,
StructArray,
+};
+use arrow_buffer::NullBuffer;
+
+fn invalid(message: impl Into<String>) -> Error {
+ Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+pub(super) fn evaluate(
+ batch: &RecordBatch,
+ root: &DataField,
+ path: &[DataField],
+ op: PredicateOperator,
+ literals: &[Datum],
+ file_fields: &[DataField],
+ scan_fields: &[DataField],
+) -> Result<BooleanArray> {
+ let leaf = path
+ .last()
+ .ok_or_else(|| invalid("Nested field path must not be empty"))?;
+ let target = crate::arrow::paimon_type_to_arrow(leaf.data_type())?;
+ let source = scan_fields
+ .iter()
+ .chain(file_fields)
+ .find(|f| f.id() == root.id());
+ let position = batch
+ .schema()
+ .fields()
+ .iter()
+ .position(|f| crate::arrow::arrow_field_id(f) == Some(root.id()))
+ .or_else(|| source.and_then(|f|
batch.schema().index_of(f.name()).ok()));
+ let values = if let Some(position) = position {
+ descend(
+ batch.column(position).clone(),
+ source.unwrap_or(root).data_type(),
+ path,
+ &target,
+ )?
+ } else if file_fields.iter().any(|f| f.id() == root.id()) {
+ return Err(invalid(format!(
+ "Nested predicate root '{}' is missing from the scanned batch",
+ root.name()
+ )));
+ } else {
+ new_null_array(&target, batch.num_rows())
+ };
+ let values = if matches!(op, PredicateOperator::IsNull |
PredicateOperator::IsNotNull)
+ || values.data_type() == &target
+ {
+ values
+ } else {
+ arrow_cast::cast_with_options(
+ values.as_ref(),
+ &target,
+ &arrow_cast::CastOptions {
+ safe: false,
+ ..Default::default()
+ },
+ )
+ .map_err(|e| {
+ invalid(format!(
+ "Cannot evolve nested predicate field '{}': {e}",
+ leaf.name()
+ ))
+ })?
+ };
+ super::evaluate_exact_leaf_predicate(&values, leaf.data_type(), op,
literals)
+ .map_err(|e| invalid(format!("Cannot evaluate nested predicate: {e}")))
+}
+
+fn descend<'a>(
+ mut values: ArrayRef,
+ mut source_type: &'a DataType,
+ path: &'a [DataField],
+ target: &arrow_schema::DataType,
+) -> Result<ArrayRef> {
+ let mut nulls = None;
+ let len = values.len();
+ for expected in path {
+ let row = values
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .ok_or_else(|| invalid("Nested predicate requires ROW input"))?;
+ nulls = NullBuffer::union(nulls.as_ref(), row.nulls());
+ let DataType::Row(logical) = source_type else {
+ return Err(invalid("Nested predicate source has a non-ROW type"));
+ };
+ let source = logical.fields().iter().find(|f| f.id() == expected.id());
+ let metadata = row
+ .fields()
+ .iter()
+ .any(|f| crate::arrow::arrow_field_id(f).is_some());
+ let index = if metadata {
+ row.fields()
+ .iter()
+ .position(|f| crate::arrow::arrow_field_id(f) ==
Some(expected.id()))
+ } else {
+ source.and_then(|f| row.fields().iter().position(|a| a.name() ==
f.name()))
+ };
+ let Some(index) = index else {
+ if source.is_some() {
+ return Err(invalid(format!(
+ "Nested predicate field '{}' is missing from the scanned
ROW",
+ expected.name()
+ )));
+ }
+ return Ok(new_null_array(target, len));
+ };
+ let next = row.column(index).clone();
+ source_type = source.map_or(expected.data_type(),
DataField::data_type);
+ values = next;
+ }
+ let nulls = NullBuffer::union(nulls.as_ref(), values.nulls());
+ let data = values
+ .to_data()
+ .into_builder()
+ .nulls(nulls)
+ .build()
+ .map_err(|e| invalid(e.to_string()))?;
+ Ok(make_array(data))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::arrow::format::FilePredicates;
+ use crate::arrow::residual::{evaluate_predicate_mask, widen_scan_fields};
+ use crate::spec::{
+ BigIntType, BinaryRowBuilder, IntType, Predicate, PredicateBuilder,
RowType,
+ };
+ use arrow_array::{Int32Array, Int64Array};
+ use arrow_schema::{DataType as ArrowType, Field, Schema};
+ use std::sync::Arc;
+
+ fn int(id: i32, name: &str) -> DataField {
+ DataField::new(id, name.into(), DataType::Int(IntType::new()))
+ }
+
+ fn row(id: i32, name: &str, fields: Vec<DataField>) -> DataField {
+ DataField::new(id, name.into(), DataType::Row(RowType::new(fields)))
+ }
+
+ fn schema() -> Vec<DataField> {
+ vec![row(
+ 10,
+ "root",
+ vec![row(20, "detail", vec![int(30, "value")])],
+ )]
+ }
+
+ fn predicate(fields: &[DataField], op: PredicateOperator, literals:
Vec<Datum>) -> Predicate {
+ let builder = PredicateBuilder::new(fields);
+ let field = builder.nested_field("root", &["detail",
"value"]).unwrap();
+ match op {
+ PredicateOperator::Eq => builder.equal(&field,
literals[0].clone()),
+ PredicateOperator::IsNull => builder.is_null(&field),
+ PredicateOperator::In => builder.is_in(&field, literals),
+ PredicateOperator::NotIn => builder.is_not_in(&field, literals),
+ _ => unreachable!("test fixture operator"),
+ }
+ .unwrap()
+ }
+
+ fn batch(fields: &[DataField], values: Vec<ArrayRef>) -> RecordBatch {
+ let schema = fields
+ .iter()
+ .map(|field| {
+ crate::arrow::paimon_type_to_arrow(field.data_type())
+ .map(|ty| Field::new(field.name(), ty, true))
+ })
+ .collect::<Result<Vec<_>>>()
+ .unwrap();
+ RecordBatch::try_new(Arc::new(Schema::new(schema)), values).unwrap()
+ }
+
+ fn mask(batch: &RecordBatch, predicate: &Predicate, fields: &[DataField])
-> Vec<bool> {
+ evaluate_predicate_mask(batch, predicate, fields, fields)
+ .unwrap()
+ .unwrap()
+ .values()
+ .iter()
+ .collect()
+ }
+
+ #[test]
+ fn nested_predicate_masks_null_ancestors_and_sliced_children() {
+ let fields = schema();
+ // NULL parents deliberately contain non-NULL physical children. Their
+ // values must never leak into equality or IS NOT NULL results.
+ let detail = StructArray::new(
+ vec![Arc::new(Field::new("value", ArrowType::Int32, true))].into(),
+ vec![Arc::new(Int32Array::from(vec![
+ Some(99),
+ Some(7),
+ Some(7),
+ None,
+ Some(7),
+ ]))],
+ Some(NullBuffer::from(vec![true, false, true, true, true])),
+ );
+ let root = StructArray::new(
+ vec![Arc::new(Field::new(
+ "detail",
+ detail.data_type().clone(),
+ true,
+ ))]
+ .into(),
+ vec![Arc::new(detail)],
+ Some(NullBuffer::from(vec![true, true, false, true, true])),
+ );
+ let batch = batch(&fields, vec![Arc::new(root)]).slice(1, 4);
+ let equal = predicate(&fields, PredicateOperator::Eq,
vec![Datum::Int(7)]);
+ assert_eq!(mask(&batch, &equal, &fields), [false, false, false, true]);
+ let is_null = predicate(&fields, PredicateOperator::IsNull, vec![]);
+ assert_eq!(mask(&batch, &is_null, &fields), [true, true, true, false]);
+ let not_in = predicate(&fields, PredicateOperator::NotIn,
vec![Datum::Int(9)]);
+ assert_eq!(mask(&batch, ¬_in, &fields), [false, false, false,
true]);
+ let empty_in = predicate(&fields, PredicateOperator::In, vec![]);
+ assert_eq!(
+ mask(&batch, &Predicate::negate(empty_in), &fields),
+ [false, false, false, true]
+ );
+ }
+
+ #[test]
+ fn nested_predicate_resolves_file_ids_after_rename_and_promotion() {
+ let old = schema();
+ let new = vec![row(
+ 10,
+ "renamed_root",
+ vec![row(
+ 20,
+ "renamed_detail",
+ vec![DataField::new(
+ 30,
+ "renamed_value".into(),
+ DataType::BigInt(BigIntType::new()),
+ )],
+ )],
+ )];
+ let detail = StructArray::new(
+ vec![Arc::new(Field::new("value", ArrowType::Int32, true))].into(),
+ vec![Arc::new(Int32Array::from(vec![Some(7), Some(8), None]))],
+ None,
+ );
+ let root = StructArray::new(
+ vec![Arc::new(Field::new(
+ "detail",
+ detail.data_type().clone(),
+ true,
+ ))]
+ .into(),
+ vec![Arc::new(detail)],
+ None,
+ );
+ let batch = batch(&old, vec![Arc::new(root)]);
+ let builder = PredicateBuilder::new(&new);
+ let field = builder
+ .nested_field("renamed_root", &["renamed_detail", "renamed_value"])
+ .unwrap();
+ let pred = builder.equal(&field, Datum::Long(7)).unwrap();
+ assert_eq!(mask(&batch, &pred, &old), [true, false, false]);
+ let pred = builder
+ .greater_than(&field, Datum::Long(i32::MAX as i64 + 1))
+ .unwrap();
+ assert_eq!(mask(&batch, &pred, &old), [false, false, false]);
+ }
+
+ #[test]
+ fn nested_predicate_treats_readded_same_name_as_missing() {
+ let old = vec![row(
+ 10,
+ "root",
+ vec![row(20, "detail", vec![int(31, "value")])],
+ )];
+ let new = schema();
+ let detail = StructArray::new(
+ vec![Arc::new(Field::new("value", ArrowType::Int32, true))].into(),
+ vec![Arc::new(Int32Array::from(vec![7, 8]))],
+ None,
+ );
+ let root = StructArray::new(
+ vec![Arc::new(Field::new(
+ "detail",
+ detail.data_type().clone(),
+ true,
+ ))]
+ .into(),
+ vec![Arc::new(detail)],
+ None,
+ );
+ let batch = batch(&old, vec![Arc::new(root)]);
+ let pred = predicate(&new, PredicateOperator::IsNull, vec![]);
+ assert_eq!(mask(&batch, &pred, &old), [true, true]);
+ let pred = predicate(&new, PredicateOperator::Eq, vec![Datum::Int(7)]);
+ assert_eq!(mask(&batch, &pred, &old), [false, false]);
+ }
+
+ #[test]
+ fn nested_predicate_prefers_arrow_ids_to_projected_positions() {
+ let fields = schema();
+ let physical = |id: i32, name: &str| {
+ Arc::new(Field::new(name, ArrowType::Int32, true).with_metadata(
+ std::collections::HashMap::from([("PARQUET:field_id".into(),
id.to_string())]),
+ ))
+ };
+ let detail = StructArray::new(
+ vec![physical(31_i32, "other"), physical(30_i32,
"renamed")].into(),
+ vec![
+ Arc::new(Int32Array::from(vec![7, 7])),
+ Arc::new(Int32Array::from(vec![8, 7])),
+ ],
+ None,
+ );
+ let root = StructArray::new(
+ vec![Arc::new(Field::new(
+ "detail",
+ detail.data_type().clone(),
+ true,
+ ))]
+ .into(),
+ vec![Arc::new(detail)],
+ None,
+ );
+ let batch = RecordBatch::try_from_iter([("root", Arc::new(root) as
ArrayRef)]).unwrap();
+ assert_eq!(
+ mask(
+ &batch,
+ &predicate(&fields, PredicateOperator::Eq,
vec![Datum::Int(7)]),
+ &fields
+ ),
+ [false, true]
+ );
+ }
+
+ #[test]
+ fn nested_predicate_widens_pruned_roots_for_unselected_filter_fields() {
+ let full = schema();
+ let projected = vec![row(10, "root", vec![row(20, "detail", vec![])])];
+ let predicate = predicate(&full, PredicateOperator::Eq,
vec![Datum::Int(7)]);
+ let predicates = FilePredicates {
+ predicates: vec![Predicate::or(vec![predicate,
Predicate::AlwaysFalse])],
+ file_fields: full.clone(),
+ row_filter_factory: None,
+ };
+ assert_eq!(widen_scan_fields(&projected, Some(&predicates)), full);
+ }
+
+ #[test]
+ fn
nested_predicate_missing_decoded_child_errors_instead_of_returning_null() {
+ let fields = schema();
+ let detail = StructArray::new_empty_fields(2, None);
+ let root = StructArray::new(
+ vec![Arc::new(Field::new(
+ "detail",
+ detail.data_type().clone(),
+ true,
+ ))]
+ .into(),
+ vec![Arc::new(detail)],
+ None,
+ );
+ let batch = RecordBatch::try_from_iter([("root", Arc::new(root) as
ArrayRef)]).unwrap();
+ assert!(evaluate_predicate_mask(
+ &batch,
+ &predicate(&fields, PredicateOperator::IsNull, vec![]),
+ &fields,
+ &fields
+ )
+ .unwrap_err()
+ .to_string()
+ .contains("missing from the scanned ROW"));
+ }
+
+ #[test]
+ fn nested_predicate_statistics_never_read_parent_bounds() {
+ struct ParentStats;
+ impl crate::predicate_stats::StatsAccessor for ParentStats {
+ fn row_count(&self) -> i64 {
+ 2
+ }
+ fn null_count(&self, _: usize) -> Option<i64> {
+ panic!("parent null count")
+ }
+ fn min_value(&self, _: usize, _: &DataType) -> Option<Datum> {
+ panic!("parent min")
+ }
+ fn max_value(&self, _: usize, _: &DataType) -> Option<Datum> {
+ panic!("parent max")
+ }
+ }
+ let fields = schema();
+ let pred = predicate(&fields, PredicateOperator::Eq,
vec![Datum::Int(7)]);
+ for pred in [
+ pred.clone(),
+ Predicate::negate(pred.clone()),
+ Predicate::and(vec![pred.clone(), Predicate::AlwaysTrue]),
+ Predicate::or(vec![pred, Predicate::AlwaysFalse]),
+ ] {
+ assert!(crate::predicate_stats::predicates_may_match_with_schema(
+ &[pred],
+ &ParentStats,
+ &[Some(0)],
+ &fields
+ ));
+ }
+ }
+
+ #[test]
+ fn nested_predicate_builder_checks_paths_literals_and_stable_identity() {
+ let fields = vec![row(
+ 10,
+ "Root",
+ vec![int(30, "a.b"), int(31, "Value"), int(32, "value")],
+ )];
+ let builder = PredicateBuilder::new(&fields);
+ assert!(builder.nested_field("Root", &[]).is_err());
+ let field = builder.nested_field("Root", &["a.b"]).unwrap();
+ let pred = builder.equal(&field, Datum::Int(7)).unwrap();
+ pred.validate_nested_fields(&fields).unwrap();
+ assert!(builder.nested_field("Root", &["a", "b"]).is_err());
+ assert!(builder.nested_field("Root", &["a.b", "child"]).is_err());
+ assert!(builder.equal(&field, Datum::Long(7)).is_err());
+ let insensitive = PredicateBuilder::new_with_case_sensitive(&fields,
false);
+ assert!(insensitive.nested_field("root", &["VALUE"]).is_err());
+ let renamed = insensitive.nested_field("root", &["A.B"]).unwrap();
+ insensitive
+ .is_null(&renamed)
+ .unwrap()
+ .validate_nested_fields(&fields)
+ .unwrap();
+ let changed = vec![row(
+ 10,
+ "Root",
+ vec![int(35, "a.b"), int(31, "Value"), int(32, "value")],
+ )];
+ assert!(pred.validate_nested_fields(&changed).is_err());
+ }
+
+ #[test]
+ fn nested_predicate_binary_rows_follow_parent_nulls() {
+ let fields = schema();
+ let pred = predicate(&fields, PredicateOperator::Eq,
vec![Datum::Int(7)]);
+ let is_null = predicate(&fields, PredicateOperator::IsNull, vec![]);
+ let mut leaf = BinaryRowBuilder::new(1);
+ leaf.write_int(0, 7);
+ let mut detail = BinaryRowBuilder::new(1);
+ detail.write_binary(0, leaf.build().data());
+ let mut row = BinaryRowBuilder::new(1);
+ row.write_binary(0, detail.build().data());
+ assert!(crate::spec::eval_row(&pred, &row.build()).unwrap());
+ let mut row = BinaryRowBuilder::new(1);
+ row.set_null_at(0);
+ assert!(crate::spec::eval_row(&is_null, &row.build()).unwrap());
+ let mut detail = BinaryRowBuilder::new(1);
+ detail.set_null_at(0);
+ let mut row = BinaryRowBuilder::new(1);
+ row.write_binary(0, detail.build().data());
+ assert!(!crate::spec::eval_row(&pred, &row.build()).unwrap());
+ // Keep an unrelated primitive column to exercise a schema with no
root.
+ let batch = RecordBatch::try_from_iter([(
+ "other",
+ Arc::new(Int64Array::from(vec![1_i64])) as ArrayRef,
+ )])
+ .unwrap();
+ assert_eq!(mask(&batch, &is_null, &[]), [true]);
+ }
+}
diff --git a/crates/paimon/src/file_index/evaluator.rs
b/crates/paimon/src/file_index/evaluator.rs
index 8fdd9bfc..39e807d5 100644
--- a/crates/paimon/src/file_index/evaluator.rs
+++ b/crates/paimon/src/file_index/evaluator.rs
@@ -106,6 +106,7 @@ fn remap_predicate(
predicate: &Predicate,
) -> Option<Predicate> {
match predicate {
+ Predicate::NestedLeaf { .. } => None,
Predicate::Leaf {
column,
index,
@@ -159,6 +160,7 @@ fn remap_predicate_exact(
predicate: &Predicate,
) -> Option<Predicate> {
match predicate {
+ Predicate::NestedLeaf { .. } => None,
Predicate::And(children) => children
.iter()
.map(|child| remap_predicate_exact(table_fields, data_fields,
child))
@@ -242,6 +244,7 @@ fn integer_datum(data_type: &DataType, value: i64) ->
Option<Datum> {
fn collect_required_columns(predicate: &Predicate, columns: &mut
HashSet<String>) {
match predicate {
+ Predicate::NestedLeaf { .. } => {}
Predicate::Leaf { column, .. } => {
columns.insert(column.clone());
}
diff --git a/crates/paimon/src/file_index/file_index_predicate.rs
b/crates/paimon/src/file_index/file_index_predicate.rs
index 2d24f2a9..f487c479 100644
--- a/crates/paimon/src/file_index/file_index_predicate.rs
+++ b/crates/paimon/src/file_index/file_index_predicate.rs
@@ -35,6 +35,7 @@ impl FileIndexPredicate {
/// Evaluates a predicate without reading data outside the supplied
indexes.
pub(crate) fn evaluate(&self, predicate: &Predicate) -> FileIndexResult {
match predicate {
+ Predicate::NestedLeaf { .. } => FileIndexResult::Remain,
Predicate::AlwaysTrue => FileIndexResult::Remain,
Predicate::AlwaysFalse => FileIndexResult::Skip,
Predicate::And(children) => self.evaluate_and(children),
diff --git a/crates/paimon/src/predicate_stats.rs
b/crates/paimon/src/predicate_stats.rs
index 49e86ac6..871705e0 100644
--- a/crates/paimon/src/predicate_stats.rs
+++ b/crates/paimon/src/predicate_stats.rs
@@ -491,6 +491,7 @@ fn predicate_may_match_with_schema<T: StatsAccessor>(
file_fields: &[DataField],
) -> bool {
match predicate {
+ Predicate::NestedLeaf { .. } => true,
Predicate::AlwaysTrue => true,
Predicate::AlwaysFalse => false,
Predicate::And(children) => children
@@ -536,6 +537,7 @@ fn predicate_must_match_with_schema<T: StatsAccessor>(
file_fields: &[DataField],
) -> bool {
match predicate {
+ Predicate::NestedLeaf { .. } => false,
Predicate::AlwaysTrue => stats.row_count() > 0,
Predicate::AlwaysFalse => false,
Predicate::And(children) => children.iter().all(|child| {
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index ba45b051..d800acca 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -134,8 +134,8 @@ pub(crate) use predicate::like_match;
pub(crate) use predicate::row_id_leaf;
pub(crate) use predicate::validate_datum_matches_type;
pub use predicate::{
- field_idx_to_partition_idx, DatePart, Datum, PadDirection, Predicate,
PredicateBuilder,
- PredicateOperator, Transform, TransformInput, TrimFlag,
+ field_idx_to_partition_idx, DatePart, Datum, FieldRef, PadDirection,
Predicate,
+ PredicateBuilder, PredicateField, PredicateOperator, Transform,
TransformInput, TrimFlag,
};
pub(crate) mod murmur_hash;
mod partition_statistics;
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index 16186f7b..dab8bb18 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -27,7 +27,15 @@ use crate::error::*;
use crate::spec::binary_row::BinaryRow;
use crate::spec::types::DataType;
use crate::spec::{is_row_tracking_column, DataField};
+use std::borrow::Cow;
use std::cmp::Ordering;
+
+mod field_ref;
+pub use field_ref::{FieldRef, PredicateField};
+
+#[cfg(test)]
+#[path = "predicate/field_ref_tests.rs"]
+mod field_ref_tests;
use std::fmt;
// ---------------------------------------------------------------------------
@@ -336,6 +344,21 @@ pub enum Predicate {
/// multiple for In/NotIn).
literals: Vec<Datum>,
},
+ /// A comparison on a field inside a ROW. Like Java's NestedFieldTransform,
+ /// this is distinct from a top-level leaf: root-column statistics and file
+ /// indexes cannot be used to prune it. Paths retain stable field
identities.
+ NestedLeaf {
+ /// Enclosing top-level ROW field in the predicate schema.
+ root: Box<DataField>,
+ /// Index of the root in the current evaluation schema.
+ index: usize,
+ /// Nonempty sequence of descendant fields, excluding the root.
+ path: Vec<DataField>,
+ /// Comparison applied to the final descendant.
+ op: PredicateOperator,
+ /// Literals typed against the final descendant.
+ literals: Vec<Datum>,
+ },
/// N-ary conjunction. Invariant: always flattened (no nested And).
And(Vec<Predicate>),
/// N-ary disjunction. Invariant: always flattened (no nested Or).
@@ -349,6 +372,29 @@ pub enum Predicate {
}
impl Predicate {
+ /// Validate a nested reference against its original schema. Keeping the
root,
+ /// every field id and each type together prevents a remap from
substituting
+ /// a same-named but unrelated field.
+ pub(crate) fn validate_nested_fields(&self, fields: &[DataField]) ->
Result<()> {
+ let Self::NestedLeaf {
+ root,
+ index,
+ path,
+ op,
+ literals,
+ } = self
+ else {
+ return Ok(());
+ };
+ let names = path.iter().map(DataField::name).collect::<Vec<_>>();
+ let builder = PredicateBuilder::new(fields);
+ let field = builder.nested_field(root.name(), &names)?;
+ let canonical = builder.leaf(&field, *op, literals.clone())?;
+ if fields.get(*index) != Some(root.as_ref()) || &canonical != self {
+ return Err(Error::DataInvalid { message: "Nested predicate does
not match the table schema; build filters with PredicateBuilder".into(),
source: None });
+ }
+ Ok(())
+ }
/// Combine predicates with AND, with recursive flattening and constant
absorption.
///
/// - `AND(p, AlwaysTrue)` → `p` (identity element filtered out)
@@ -432,6 +478,27 @@ impl Predicate {
Predicate::Not(inner) => *inner,
Predicate::AlwaysTrue => Predicate::AlwaysFalse,
Predicate::AlwaysFalse => Predicate::AlwaysTrue,
+ Predicate::NestedLeaf {
+ root,
+ index,
+ path,
+ op,
+ literals,
+ } if literals.is_empty()
+ && matches!(op, PredicateOperator::In |
PredicateOperator::NotIn) =>
+ {
+ Predicate::NestedLeaf {
+ root,
+ index,
+ path,
+ op: if op == PredicateOperator::In {
+ PredicateOperator::NotIn
+ } else {
+ PredicateOperator::In
+ },
+ literals,
+ }
+ }
Predicate::Leaf {
column,
index,
@@ -483,6 +550,7 @@ impl Predicate {
/// `mapping` is the output of `field_idx_to_partition_idx`.
pub(crate) fn remap_field_index(&self, mapping: &[Option<usize>]) ->
Option<Predicate> {
match self {
+ Predicate::NestedLeaf { .. } => None,
Predicate::Leaf {
column,
index,
@@ -531,6 +599,7 @@ 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::NestedLeaf { .. } => false,
Predicate::Leaf { column, index, .. } => {
!is_row_tracking_column(column) &&
mapping.get(*index).is_some_and(Option::is_some)
}
@@ -557,6 +626,7 @@ impl Predicate {
mapping: &[Option<usize>],
) -> Option<Predicate> {
match self {
+ Predicate::NestedLeaf { .. } => None,
Predicate::Leaf {
column,
index,
@@ -607,6 +677,26 @@ impl Predicate {
impl fmt::Display for Predicate {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
+ Self::NestedLeaf {
+ root,
+ path,
+ op,
+ literals,
+ ..
+ } => {
+ let name = std::iter::once(root.name())
+ .chain(path.iter().map(DataField::name))
+ .collect::<Vec<_>>()
+ .join(".");
+ Self::Leaf {
+ column: name,
+ index: 0,
+ data_type: path.last().unwrap().data_type().clone(),
+ op: *op,
+ literals: literals.clone(),
+ }
+ .fmt(f)
+ }
Self::Leaf {
column,
op,
@@ -682,6 +772,9 @@ impl Predicate {
/// predicate.
pub fn collect_leaf_field_indices(&self, out: &mut
std::collections::HashSet<usize>) {
match self {
+ Predicate::NestedLeaf { index, .. } => {
+ out.insert(*index);
+ }
Predicate::Leaf { index, .. } => {
out.insert(*index);
}
@@ -704,6 +797,7 @@ impl Predicate {
/// indexed by partition field.
pub fn to_rest_json(&self) -> Option<serde_json::Value> {
match self {
+ Predicate::NestedLeaf { .. } => None,
Predicate::AlwaysTrue => Some(rest_constant_leaf("TRUE")),
Predicate::AlwaysFalse => Some(rest_constant_leaf("FALSE")),
Predicate::And(children) | Predicate::Or(children) => {
@@ -1645,7 +1739,8 @@ fn check_input_type(input: &serde_json::Value, integer:
bool) -> Result<()> {
// PredicateBuilder
// ---------------------------------------------------------------------------
-/// Builds `Predicate` nodes from field names and typed literals.
+/// Builds `Predicate` nodes from field names or resolved field references and
+/// typed literals. Field selection is independent of the comparison operator.
///
/// Stores schema field metadata and validates column references in builder
/// methods. Unknown column names cause immediate errors (fail-fast).
@@ -1653,8 +1748,7 @@ fn check_input_type(input: &serde_json::Value, integer:
bool) -> Result<()> {
/// Reference: Java `PredicateBuilder` — but uses field names instead of
indices
/// for a more ergonomic Rust API.
pub struct PredicateBuilder {
- field_names: Vec<String>,
- field_types: Vec<DataType>,
+ fields: Vec<DataField>,
case_sensitive: bool,
}
@@ -1670,45 +1764,135 @@ impl PredicateBuilder {
/// ambiguous (case-colliding) schema errors. Infallible.
pub fn new_with_case_sensitive(fields: &[DataField], case_sensitive: bool)
-> Self {
Self {
- field_names: fields.iter().map(|f| f.name().to_string()).collect(),
- field_types: fields.iter().map(|f|
f.data_type().clone()).collect(),
+ fields: fields.to_vec(),
case_sensitive,
}
}
// -- comparison operators --
- pub fn equal(&self, field: &str, literal: Datum) -> Result<Predicate> {
+ pub fn equal<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literal: Datum,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::Eq, vec![literal])
}
- pub fn not_equal(&self, field: &str, literal: Datum) -> Result<Predicate> {
+ /// Resolve a top-level field for reuse with ordinary predicate operators.
+ pub fn field(&self, name: &str) -> Result<FieldRef> {
+ let index = self.resolve_field(name)?;
+ Ok(FieldRef {
+ root: Box::new(self.fields[index].clone()),
+ index,
+ path: vec![],
+ })
+ }
+
+ /// Resolve a nested ROW input independently of a predicate operator.
+ /// Path components are separate names, so dots in real column names remain
+ /// unambiguous. NULL at any parent yields NULL during evaluation.
+ ///
+ /// ```
+ /// # use paimon::spec::{DataField, DataType, Datum, IntType, RowType,
PredicateBuilder};
+ /// let fields = vec![DataField::new(0, "profile".into(), DataType::Row(
+ /// RowType::new(vec![DataField::new(1, "score".into(),
DataType::Int(IntType::new()))])
+ /// ))];
+ /// let builder = PredicateBuilder::new(&fields);
+ /// let score: paimon::spec::FieldRef = builder.nested_field("profile",
&["score"])?;
+ /// let equals = builder.equal(&score, Datum::Int(7))?;
+ /// let present = builder.is_not_null(&score)?;
+ /// # Ok::<(), paimon::Error>(())
+ /// ```
+ pub fn nested_field(&self, root: &str, path: &[&str]) -> Result<FieldRef> {
+ let index = self.resolve_field(root)?;
+ if path.is_empty() {
+ return Err(Error::ConfigInvalid {
+ message: "Nested field path must not be empty".into(),
+ });
+ }
+ let root = self.fields[index].clone();
+ let mut current = root.data_type();
+ let mut resolved = Vec::with_capacity(path.len());
+ for name in path {
+ let DataType::Row(row) = current else {
+ return Err(Error::ConfigInvalid {
+ message: format!("Nested field path descends into non-ROW
type {current:?}"),
+ });
+ };
+ let field = row
+ .fields()
+ .iter()
+ .filter(|f| {
+ if self.case_sensitive {
+ f.name() == *name
+ } else {
+ f.name().eq_ignore_ascii_case(name)
+ }
+ })
+ .collect::<Vec<_>>();
+ if field.len() != 1 {
+ return Err(Error::ConfigInvalid {
+ message: format!("Nested field '{name}' is missing or
ambiguous"),
+ });
+ }
+ resolved.push(field[0].clone());
+ current = field[0].data_type();
+ }
+ Ok(FieldRef {
+ root: Box::new(root),
+ index,
+ path: resolved,
+ })
+ }
+
+ pub fn not_equal<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literal: Datum,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::NotEq, vec![literal])
}
- pub fn less_than(&self, field: &str, literal: Datum) -> Result<Predicate> {
+ pub fn less_than<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literal: Datum,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::Lt, vec![literal])
}
- pub fn less_or_equal(&self, field: &str, literal: Datum) ->
Result<Predicate> {
+ pub fn less_or_equal<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literal: Datum,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::LtEq, vec![literal])
}
- pub fn greater_than(&self, field: &str, literal: Datum) ->
Result<Predicate> {
+ pub fn greater_than<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literal: Datum,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::Gt, vec![literal])
}
- pub fn greater_or_equal(&self, field: &str, literal: Datum) ->
Result<Predicate> {
+ pub fn greater_or_equal<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literal: Datum,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::GtEq, vec![literal])
}
// -- null operators --
- pub fn is_null(&self, field: &str) -> Result<Predicate> {
+ pub fn is_null<'a>(&self, field: impl Into<PredicateField<'a>>) ->
Result<Predicate> {
self.leaf(field, PredicateOperator::IsNull, vec![])
}
- pub fn is_not_null(&self, field: &str) -> Result<Predicate> {
+ pub fn is_not_null<'a>(&self, field: impl Into<PredicateField<'a>>) ->
Result<Predicate> {
self.leaf(field, PredicateOperator::IsNotNull, vec![])
}
@@ -1717,8 +1901,8 @@ impl PredicateBuilder {
let predicates: Vec<Predicate> = fields
.iter()
.map(|(name, value)| match value {
- Some(v) => self.equal(name, v.clone()),
- None => self.is_null(name),
+ Some(v) => self.equal(*name, v.clone()),
+ None => self.is_null(*name),
})
.collect::<Result<Vec<_>>>()?;
Ok(Predicate::and(predicates))
@@ -1726,58 +1910,92 @@ impl PredicateBuilder {
// -- set operators --
- pub fn is_in(&self, field: &str, literals: Vec<Datum>) ->
Result<Predicate> {
+ pub fn is_in<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literals: Vec<Datum>,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::In, literals)
}
- pub fn is_not_in(&self, field: &str, literals: Vec<Datum>) ->
Result<Predicate> {
+ pub fn is_not_in<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literals: Vec<Datum>,
+ ) -> Result<Predicate> {
self.leaf(field, PredicateOperator::NotIn, literals)
}
/// Java IN ignores NULL literals; an empty/all-NULL list matches nothing.
- pub fn is_in_with_nulls(&self, field: &str, literals: Vec<Option<Datum>>)
-> Result<Predicate> {
+ pub fn is_in_with_nulls<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ literals: Vec<Option<Datum>>,
+ ) -> Result<Predicate> {
self.is_in(field, literals.into_iter().flatten().collect())
}
/// Java NOT IN fails every row if any literal is NULL.
- pub fn is_not_in_with_nulls(
+ pub fn is_not_in_with_nulls<'a>(
&self,
- field: &str,
+ field: impl Into<PredicateField<'a>>,
literals: Vec<Option<Datum>>,
) -> Result<Predicate> {
- if literals.iter().any(Option::is_none) {
- self.resolve_field(field)?;
- return Ok(Predicate::AlwaysFalse);
- }
- self.is_not_in(field, literals.into_iter().flatten().collect())
+ let contains_null = literals.iter().any(Option::is_none);
+ let predicate = self.is_not_in(field,
literals.into_iter().flatten().collect())?;
+ Ok(if contains_null {
+ Predicate::AlwaysFalse
+ } else {
+ predicate
+ })
}
// -- string operators --
/// `field LIKE 'pat%'` shape. Retains an empty pattern as a string leaf,
/// matching Java. Non-string `pattern` → [`Error::ConfigInvalid`].
- pub fn starts_with(&self, field: &str, pattern: Datum) ->
Result<Predicate> {
+ pub fn starts_with<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ pattern: Datum,
+ ) -> Result<Predicate> {
self.string_leaf(field, PredicateOperator::StartsWith, pattern)
}
/// `field LIKE '%pat'` shape, including an empty pattern.
- pub fn ends_with(&self, field: &str, pattern: Datum) -> Result<Predicate> {
+ pub fn ends_with<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ pattern: Datum,
+ ) -> Result<Predicate> {
self.string_leaf(field, PredicateOperator::EndsWith, pattern)
}
/// `field LIKE '%pat%'` shape, including an empty pattern.
- pub fn contains(&self, field: &str, pattern: Datum) -> Result<Predicate> {
+ pub fn contains<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ pattern: Datum,
+ ) -> Result<Predicate> {
self.string_leaf(field, PredicateOperator::Contains, pattern)
}
/// Build an element-membership predicate over an ARRAY column.
- pub fn array_contains(&self, field: &str, element: Datum) ->
Result<Predicate> {
- self.array_leaf(field, PredicateOperator::ArrayContains, vec![element])
+ pub fn array_contains<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ element: Datum,
+ ) -> Result<Predicate> {
+ self.leaf(field, PredicateOperator::ArrayContains, vec![element])
}
/// Build an any-element membership predicate over an ARRAY column.
- pub fn arrays_overlap(&self, field: &str, elements: Vec<Datum>) ->
Result<Predicate> {
- self.array_leaf(field, PredicateOperator::ArraysOverlap, elements)
+ pub fn arrays_overlap<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ elements: Vec<Datum>,
+ ) -> Result<Predicate> {
+ self.leaf(field, PredicateOperator::ArraysOverlap, elements)
}
/// Build an all-elements membership predicate over an ARRAY column.
@@ -1785,8 +2003,12 @@ impl PredicateBuilder {
/// An empty literal list is retained as a leaf: it matches non-null
arrays,
/// while the multivalue index must fall back because it cannot distinguish
/// null arrays from empty arrays.
- pub fn array_contains_all(&self, field: &str, elements: Vec<Datum>) ->
Result<Predicate> {
- self.array_leaf(field, PredicateOperator::ArrayContainsAll, elements)
+ pub fn array_contains_all<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ elements: Vec<Datum>,
+ ) -> Result<Predicate> {
+ self.leaf(field, PredicateOperator::ArrayContainsAll, elements)
}
/// `field LIKE '<pattern>'` with optional `escape` character (default
`\`).
@@ -1801,7 +2023,12 @@ impl PredicateBuilder {
/// `escape == None` defaults to `\`. Any other ESCAPE character is
/// rejected with [`Error::ConfigInvalid`] (the DataFusion translator turns
/// that into a fall-open).
- pub fn like(&self, field: &str, pattern: Datum, escape: Option<char>) ->
Result<Predicate> {
+ pub fn like<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ pattern: Datum,
+ escape: Option<char>,
+ ) -> Result<Predicate> {
let pattern_str = match &pattern {
Datum::String(s) => s.clone(),
other => {
@@ -1833,22 +2060,26 @@ impl PredicateBuilder {
/// `field BETWEEN low AND high`. SQL semantics: inclusive on both ends.
/// `low > high` short-circuits to [`Predicate::AlwaysFalse`] (no value can
/// be between an empty range).
- pub fn between(&self, field: &str, low: Datum, high: Datum) ->
Result<Predicate> {
- if Self::low_strictly_above_high(&low, &high) {
- return Ok(Predicate::AlwaysFalse);
- }
- self.leaf(field, PredicateOperator::Between, vec![low, high])
+ pub fn between<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ low: Datum,
+ high: Datum,
+ ) -> Result<Predicate> {
+ self.range_leaf(field, PredicateOperator::Between, low, high)
}
/// `field NOT BETWEEN low AND high`. SQL three-valued logic: NULL value
/// → NULL → treated as false, matching the existing `NotEq` evaluator.
/// `low > high` short-circuits to `IsNotNull(field)` (every non-null value
/// is "not between" an empty range).
- pub fn not_between(&self, field: &str, low: Datum, high: Datum) ->
Result<Predicate> {
- if Self::low_strictly_above_high(&low, &high) {
- return self.is_not_null(field);
- }
- self.leaf(field, PredicateOperator::NotBetween, vec![low, high])
+ pub fn not_between<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ low: Datum,
+ high: Datum,
+ ) -> Result<Predicate> {
+ self.range_leaf(field, PredicateOperator::NotBetween, low, high)
}
fn low_strictly_above_high(low: &Datum, high: &Datum) -> bool {
@@ -1858,7 +2089,12 @@ impl PredicateBuilder {
/// Shared literal-type guard for the three string operators. Preserve
/// empty string leaves: search refinement depends on the original
operator,
/// even when row evaluation is equivalent to `IsNotNull`.
- fn string_leaf(&self, field: &str, op: PredicateOperator, pattern: Datum)
-> Result<Predicate> {
+ fn string_leaf<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ op: PredicateOperator,
+ pattern: Datum,
+ ) -> Result<Predicate> {
match &pattern {
Datum::String(_) => {}
other => {
@@ -1870,91 +2106,118 @@ impl PredicateBuilder {
self.leaf(field, op, vec![pattern])
}
- fn array_leaf(
+ fn range_leaf<'a>(
&self,
- field: &str,
+ field: impl Into<PredicateField<'a>>,
op: PredicateOperator,
- literals: Vec<Datum>,
+ low: Datum,
+ high: Datum,
) -> Result<Predicate> {
- let (index, canonical, data_type) = self.resolve_field(field)?;
- let DataType::Array(array_type) = &data_type else {
- return Err(Error::ConfigInvalid {
- message: format!("{op} requires an ARRAY field, got
{data_type:?}"),
- });
- };
- Self::validate_literal_count(op, &literals)?;
- for literal in &literals {
- validate_datum_matches_type(literal, array_type.element_type())?;
- }
- Ok(Predicate::Leaf {
- column: canonical,
- index,
- data_type,
- op,
- literals,
+ let field = self.resolve_input(field)?;
+ let reversed = Self::low_strictly_above_high(&low, &high);
+ let predicate = Self::resolved_leaf(&field, op, vec![low, high])?;
+ Ok(if reversed {
+ if op == PredicateOperator::Between {
+ Predicate::AlwaysFalse
+ } else {
+ field.predicate(PredicateOperator::IsNotNull, vec![])
+ }
+ } else {
+ predicate
})
}
// -- internal --
- /// Resolve field name to index + type, validate literals, and build a
leaf predicate.
- fn leaf(&self, field: &str, op: PredicateOperator, literals: Vec<Datum>)
-> Result<Predicate> {
- let (index, canonical, data_type) = self.resolve_field(field)?;
+ fn leaf<'a>(
+ &self,
+ field: impl Into<PredicateField<'a>>,
+ op: PredicateOperator,
+ literals: Vec<Datum>,
+ ) -> Result<Predicate> {
+ let field = self.resolve_input(field)?;
+ Self::resolved_leaf(&field, op, literals)
+ }
+
+ fn resolved_leaf(
+ field: &FieldRef,
+ op: PredicateOperator,
+ literals: Vec<Datum>,
+ ) -> Result<Predicate> {
+ let data_type = field.data_type();
+ let literal_type = if matches!(
+ op,
+ PredicateOperator::ArrayContains
+ | PredicateOperator::ArraysOverlap
+ | PredicateOperator::ArrayContainsAll
+ ) {
+ let DataType::Array(array_type) = data_type else {
+ return Err(Error::ConfigInvalid {
+ message: format!("{op} requires an ARRAY field, got
{data_type:?}"),
+ });
+ };
+ array_type.element_type()
+ } else {
+ data_type
+ };
Self::validate_literal_count(op, &literals)?;
- for lit in &literals {
- validate_datum_matches_type(lit, &data_type)?;
+ for literal in &literals {
+ validate_datum_matches_type(literal, literal_type)?;
+ }
+ Ok(field.predicate(op, literals))
+ }
+
+ fn resolve_input<'a>(&self, input: impl Into<PredicateField<'a>>) ->
Result<Cow<'a, FieldRef>> {
+ match input.into().0 {
+ field_ref::FieldInput::Name(name) =>
self.field(&name).map(Cow::Owned),
+ field_ref::FieldInput::Reference(field) => {
+ // A resolved operand may be reused only with the same root
+ // identity, position and schema. Validate before any operator
+ // optimization can turn the predicate into a constant.
+ if self.fields.get(field.index) != Some(field.root.as_ref()) {
+ return Err(Error::ConfigInvalid {
+ message: "Predicate field reference does not match the
builder schema"
+ .into(),
+ });
+ }
+ Ok(field)
+ }
}
- Ok(Predicate::Leaf {
- column: canonical,
- index,
- data_type,
- op,
- literals,
- })
}
- /// Look up a field name, returning its (index, DataType) or an error.
- /// Resolve a column name to `(index, canonical_name, data_type)`.
+ /// Resolve a column name to its canonical schema position.
///
/// The canonical name is the schema's stored field name for the matched
/// index. It is stored in the resulting [`Predicate::Leaf`] so downstream
/// name-based lookups (e.g. global-index pruning) match the schema even
/// when the caller passed a different casing. Case-insensitive matching
/// errors on an ambiguous (case-colliding) schema.
- fn resolve_field(&self, field: &str) -> Result<(usize, String, DataType)> {
+ fn resolve_field(&self, field: &str) -> Result<usize> {
let index = if self.case_sensitive {
- self.field_names.iter().position(|n| n == field)
+ self.fields.iter().position(|f| f.name() == field)
} else {
let mut hits = self
- .field_names
+ .fields
.iter()
.enumerate()
- .filter(|(_, n)| n.eq_ignore_ascii_case(field));
+ .filter(|(_, f)| f.name().eq_ignore_ascii_case(field));
let first = hits.next();
if first.is_some() && hits.next().is_some() {
return Err(Error::ConfigInvalid {
message: format!(
"Ambiguous column '{field}' matches multiple schema
fields case-insensitively: {:?}",
- self.field_names
+ self.fields
),
});
}
first.map(|(i, _)| i)
};
- index
- .map(|idx| {
- (
- idx,
- self.field_names[idx].clone(),
- self.field_types[idx].clone(),
- )
- })
- .ok_or_else(|| Error::ConfigInvalid {
- message: format!(
- "Column '{}' not found in schema fields {:?}",
- field, self.field_names
- ),
- })
+ index.ok_or_else(|| Error::ConfigInvalid {
+ message: format!(
+ "Column '{}' not found in schema fields {:?}",
+ field, self.fields
+ ),
+ })
}
/// Validate that the number of literals matches the operator's
expectation.
@@ -2135,6 +2398,43 @@ pub(crate) fn extract_datum(
/// SQL null semantics: null compared to any value yields `false`.
pub(crate) fn eval_row(predicate: &Predicate, row: &BinaryRow) -> Result<bool>
{
match predicate {
+ Predicate::NestedLeaf {
+ root,
+ index,
+ path,
+ op,
+ literals,
+ } => {
+ let mut current = row.clone();
+ let mut position = *index;
+ let mut data_type = root.data_type();
+ for field in path {
+ if current.is_null_at(position) {
+ return Ok(eval_leaf(*op, None, literals));
+ }
+ let DataType::Row(row_type) = data_type else {
+ return Err(Error::DataInvalid {
+ message: "Nested predicate requires ROW input".into(),
+ source: None,
+ });
+ };
+ current = BinaryRow::from_bytes(
+ row_type.fields().len() as i32,
+ current.get_binary(position)?.to_vec(),
+ );
+ position = row_type
+ .fields()
+ .iter()
+ .position(|f| f.id() == field.id())
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Nested field '{}' changed identity",
field.name()),
+ source: None,
+ })?;
+ data_type = field.data_type();
+ }
+ let value = extract_datum(¤t, position, data_type)?;
+ Ok(eval_leaf(*op, value.as_ref(), literals))
+ }
Predicate::AlwaysTrue => Ok(true),
Predicate::AlwaysFalse => Ok(false),
Predicate::And(children) => {
diff --git a/crates/paimon/src/spec/predicate/field_ref.rs
b/crates/paimon/src/spec/predicate/field_ref.rs
new file mode 100644
index 00000000..0342e70c
--- /dev/null
+++ b/crates/paimon/src/spec/predicate/field_ref.rs
@@ -0,0 +1,106 @@
+// 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.
+
+//! Resolved predicate operands and Rust's name/reference argument adapter.
+
+use super::Predicate;
+use crate::spec::{DataField, DataType, Datum, PredicateOperator};
+use std::borrow::Cow;
+
+/// A schema-bound input to predicate comparisons, including a field inside a
ROW.
+///
+/// Construct with [`super::PredicateBuilder::field`] or
+/// [`super::PredicateBuilder::nested_field`], then reuse it with the ordinary
+/// comparison, null, set, string and range builder methods. Like Java's
+/// `NestedFieldTransform`, it describes the value independently of the
operator.
+/// Its immutable field identities are validated by the receiving builder.
+#[derive(Debug, Clone, PartialEq)]
+pub struct FieldRef {
+ pub(super) root: Box<DataField>,
+ pub(super) index: usize,
+ pub(super) path: Vec<DataField>,
+}
+
+impl FieldRef {
+ /// Type of the selected value, rather than the enclosing ROW's type.
+ pub fn data_type(&self) -> &DataType {
+ self.path.last().unwrap_or(&self.root).data_type()
+ }
+
+ pub(super) fn predicate(&self, op: PredicateOperator, literals:
Vec<Datum>) -> Predicate {
+ if self.path.is_empty() {
+ Predicate::Leaf {
+ column: self.root.name().into(),
+ index: self.index,
+ data_type: self.root.data_type().clone(),
+ op,
+ literals,
+ }
+ } else {
+ Predicate::NestedLeaf {
+ root: self.root.clone(),
+ index: self.index,
+ path: self.path.clone(),
+ op,
+ literals,
+ }
+ }
+ }
+}
+
+/// A column name or resolved [`FieldRef`] accepted by
[`super::PredicateBuilder`].
+///
+/// Builder methods accept `impl Into<PredicateField>` so existing `&str` and
+/// `String` names, and either owned or borrowed field references, share the
same
+/// operators and literal validation. Callers do not need to construct this
+/// adapter explicitly.
+pub struct PredicateField<'a>(pub(super) FieldInput<'a>);
+
+pub(super) enum FieldInput<'a> {
+ Name(Cow<'a, str>),
+ Reference(Cow<'a, FieldRef>),
+}
+
+impl<'a> From<&'a str> for PredicateField<'a> {
+ fn from(name: &'a str) -> Self {
+ Self(FieldInput::Name(Cow::Borrowed(name)))
+ }
+}
+
+impl<'a> From<&'a String> for PredicateField<'a> {
+ fn from(name: &'a String) -> Self {
+ Self::from(name.as_str())
+ }
+}
+
+impl From<String> for PredicateField<'_> {
+ fn from(name: String) -> Self {
+ Self(FieldInput::Name(Cow::Owned(name)))
+ }
+}
+
+impl<'a> From<&'a FieldRef> for PredicateField<'a> {
+ fn from(field: &'a FieldRef) -> Self {
+ Self(FieldInput::Reference(Cow::Borrowed(field)))
+ }
+}
+
+impl From<FieldRef> for PredicateField<'_> {
+ fn from(field: FieldRef) -> Self {
+ Self(FieldInput::Reference(Cow::Owned(field)))
+ }
+}
diff --git a/crates/paimon/src/spec/predicate/field_ref_tests.rs
b/crates/paimon/src/spec/predicate/field_ref_tests.rs
new file mode 100644
index 00000000..15578a10
--- /dev/null
+++ b/crates/paimon/src/spec/predicate/field_ref_tests.rs
@@ -0,0 +1,380 @@
+// 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.
+
+use super::{Datum, Predicate, PredicateBuilder, PredicateOperator};
+use crate::spec::{ArrayType, DataField, DataType, IntType, RowType,
VarCharType};
+
+fn fields(root_id: i32, score_id: i32, score_type: DataType) -> Vec<DataField>
{
+ vec![
+ DataField::new(1, "id".into(), DataType::Int(IntType::new())),
+ DataField::new(
+ root_id,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ 11,
+ "detail".into(),
+ DataType::Row(RowType::new(vec![
+ DataField::new(score_id, "score".into(), score_type),
+ DataField::new(
+ 13,
+ "label".into(),
+ DataType::VarChar(VarCharType::new(100).unwrap()),
+ ),
+ DataField::new(
+ 14,
+ "values".into(),
+
DataType::Array(ArrayType::new(DataType::Int(IntType::new()))),
+ ),
+ ])),
+ )])),
+ ),
+ ]
+}
+
+fn schema() -> Vec<DataField> {
+ fields(10, 12, DataType::Int(IntType::new()))
+}
+
+fn assert_nested(predicate: Predicate, expected: PredicateOperator, literals:
&[Datum]) {
+ match predicate {
+ Predicate::NestedLeaf {
+ root,
+ index,
+ path,
+ op,
+ literals: actual,
+ } => {
+ assert_eq!(root.id(), 10);
+ assert_eq!(index, 1);
+ assert_eq!(path.iter().map(DataField::id).collect::<Vec<_>>()[0],
11);
+ assert_eq!(path.len(), 2);
+ assert_eq!(op, expected);
+ assert_eq!(actual, literals);
+ }
+ other => panic!("expected a nested comparison, got {other:?}"),
+ }
+}
+
+#[test]
+fn field_references_preserve_string_callers_and_can_be_reused() {
+ let builder = PredicateBuilder::new(&schema());
+ let name = String::from("id");
+ let id = builder.field(&name).unwrap();
+ let expected = builder.equal("id", Datum::Int(7)).unwrap();
+ assert_eq!(
+ builder.equal(name.clone(), Datum::Int(7)).unwrap(),
+ expected
+ );
+ assert_eq!(builder.equal(&name, Datum::Int(7)).unwrap(), expected);
+ assert_eq!(builder.equal(&id, Datum::Int(7)).unwrap(), expected);
+ assert_eq!(builder.equal(id, Datum::Int(7)).unwrap(), expected);
+
+ let score = builder
+ .nested_field("profile", &["detail", "score"])
+ .unwrap();
+ assert_eq!(score.data_type(), &DataType::Int(IntType::new()));
+ let same_schema = PredicateBuilder::new(&schema());
+ assert_eq!(
+ same_schema.equal(&score, Datum::Int(7)).unwrap(),
+ builder.equal(score, Datum::Int(7)).unwrap()
+ );
+}
+
+#[test]
+fn nested_field_uses_all_ordinary_comparisons_and_null_checks() {
+ let builder = PredicateBuilder::new(&schema());
+ let score = builder
+ .nested_field("profile", &["detail", "score"])
+ .unwrap();
+ let literal = Datum::Int(7);
+ for (op, predicate) in [
+ (
+ PredicateOperator::Eq,
+ builder.equal(&score, literal.clone()),
+ ),
+ (
+ PredicateOperator::NotEq,
+ builder.not_equal(&score, literal.clone()),
+ ),
+ (
+ PredicateOperator::Lt,
+ builder.less_than(&score, literal.clone()),
+ ),
+ (
+ PredicateOperator::LtEq,
+ builder.less_or_equal(&score, literal.clone()),
+ ),
+ (
+ PredicateOperator::Gt,
+ builder.greater_than(&score, literal.clone()),
+ ),
+ (
+ PredicateOperator::GtEq,
+ builder.greater_or_equal(&score, literal.clone()),
+ ),
+ ] {
+ assert_nested(predicate.unwrap(), op, std::slice::from_ref(&literal));
+ }
+ assert_nested(
+ builder.is_null(&score).unwrap(),
+ PredicateOperator::IsNull,
+ &[],
+ );
+ assert_nested(
+ builder.is_not_null(score).unwrap(),
+ PredicateOperator::IsNotNull,
+ &[],
+ );
+}
+
+#[test]
+fn nested_set_and_range_helpers_keep_java_null_semantics() {
+ let builder = PredicateBuilder::new(&schema());
+ let score = builder
+ .nested_field("profile", &["detail", "score"])
+ .unwrap();
+ let values = vec![Datum::Int(1), Datum::Int(7)];
+ assert_nested(
+ builder.is_in(&score, values.clone()).unwrap(),
+ PredicateOperator::In,
+ &values,
+ );
+ assert_nested(
+ builder.is_not_in(&score, values.clone()).unwrap(),
+ PredicateOperator::NotIn,
+ &values,
+ );
+ assert_nested(
+ builder
+ .is_in_with_nulls(&score, vec![None, Some(Datum::Int(7))])
+ .unwrap(),
+ PredicateOperator::In,
+ &[Datum::Int(7)],
+ );
+ assert_nested(
+ builder.is_in_with_nulls(&score, vec![None]).unwrap(),
+ PredicateOperator::In,
+ &[],
+ );
+ assert!(matches!(
+ builder
+ .is_not_in_with_nulls(&score, vec![Some(Datum::Int(7)), None])
+ .unwrap(),
+ Predicate::AlwaysFalse
+ ));
+ assert_nested(
+ builder
+ .between(&score, values[0].clone(), values[1].clone())
+ .unwrap(),
+ PredicateOperator::Between,
+ &values,
+ );
+ assert_nested(
+ builder
+ .not_between(&score, values[0].clone(), values[1].clone())
+ .unwrap(),
+ PredicateOperator::NotBetween,
+ &values,
+ );
+ assert!(matches!(
+ builder
+ .between(&score, Datum::Int(7), Datum::Int(1))
+ .unwrap(),
+ Predicate::AlwaysFalse
+ ));
+ assert_nested(
+ builder
+ .not_between(&score, Datum::Int(7), Datum::Int(1))
+ .unwrap(),
+ PredicateOperator::IsNotNull,
+ &[],
+ );
+}
+
+#[test]
+fn nested_like_optimization_and_string_helpers_retain_the_reference() {
+ let builder = PredicateBuilder::new(&schema());
+ let label = builder
+ .nested_field("profile", &["detail", "label"])
+ .unwrap();
+ for (pattern, op, literal) in [
+ ("a", PredicateOperator::Eq, "a"),
+ ("a%", PredicateOperator::StartsWith, "a"),
+ ("%a", PredicateOperator::EndsWith, "a"),
+ ("%a%", PredicateOperator::Contains, "a"),
+ ("a_b", PredicateOperator::Like, "a_b"),
+ ] {
+ assert_nested(
+ builder
+ .like(&label, Datum::String(pattern.into()), None)
+ .unwrap(),
+ op,
+ &[Datum::String(literal.into())],
+ );
+ }
+ for (op, predicate) in [
+ (
+ PredicateOperator::StartsWith,
+ builder.starts_with(&label, Datum::String("".into())),
+ ),
+ (
+ PredicateOperator::EndsWith,
+ builder.ends_with(&label, Datum::String("".into())),
+ ),
+ (
+ PredicateOperator::Contains,
+ builder.contains(&label, Datum::String("".into())),
+ ),
+ ] {
+ assert_nested(predicate.unwrap(), op, &[Datum::String("".into())]);
+ }
+ assert!(builder
+ .like(&label, Datum::String("a%".into()), Some('!'))
+ .is_err());
+ assert!(builder.contains(&label, Datum::Int(7)).is_err());
+ let score = builder
+ .nested_field("profile", &["detail", "score"])
+ .unwrap();
+ assert!(builder.contains(score, Datum::String("7".into())).is_err());
+}
+
+#[test]
+fn nested_array_helpers_validate_the_element_type() {
+ let builder = PredicateBuilder::new(&schema());
+ let values = builder
+ .nested_field("profile", &["detail", "values"])
+ .unwrap();
+ assert_nested(
+ builder.array_contains(&values, Datum::Int(7)).unwrap(),
+ PredicateOperator::ArrayContains,
+ &[Datum::Int(7)],
+ );
+ assert_nested(
+ builder
+ .arrays_overlap(&values, vec![Datum::Int(1), Datum::Int(7)])
+ .unwrap(),
+ PredicateOperator::ArraysOverlap,
+ &[Datum::Int(1), Datum::Int(7)],
+ );
+ assert_nested(
+ builder.array_contains_all(&values, vec![]).unwrap(),
+ PredicateOperator::ArrayContainsAll,
+ &[],
+ );
+ assert!(builder
+ .array_contains(&values, Datum::String("7".into()))
+ .is_err());
+ assert!(builder
+ .arrays_overlap(&values, vec![Datum::String("7".into())])
+ .is_err());
+ assert!(builder
+ .array_contains_all(values, vec![Datum::String("7".into())])
+ .is_err());
+ let score = builder
+ .nested_field("profile", &["detail", "score"])
+ .unwrap();
+ assert!(builder.array_contains(score, Datum::Int(7)).is_err());
+}
+
+#[test]
+fn
field_references_reject_changed_identity_and_position_before_constant_folding()
{
+ let builder = PredicateBuilder::new(&schema());
+ let score = builder
+ .nested_field("profile", &["detail", "score"])
+ .unwrap();
+ let mut reordered = schema();
+ reordered.swap(0, 1);
+ for changed in [
+ fields(100, 12, DataType::Int(IntType::new())),
+ fields(10, 120, DataType::Int(IntType::new())),
+ fields(10, 12, DataType::VarChar(VarCharType::new(100).unwrap())),
+ reordered,
+ ] {
+ let other = PredicateBuilder::new(&changed);
+ assert!(other.equal(&score, Datum::Int(7)).is_err());
+ assert!(other.is_null(&score).is_err());
+ assert!(other.is_in(&score, vec![]).is_err());
+ assert!(other.is_in_with_nulls(&score, vec![None]).is_err());
+ assert!(other.is_not_in_with_nulls(&score, vec![None]).is_err());
+ assert!(other.between(&score, Datum::Int(7), Datum::Int(1)).is_err());
+ assert!(other
+ .not_between(&score, Datum::Int(7), Datum::Int(1))
+ .is_err());
+ }
+ let id = builder.field("id").unwrap();
+ let mut changed = schema();
+ changed[0] = DataField::new(100, "id".into(),
DataType::Int(IntType::new()));
+ assert!(PredicateBuilder::new(&changed)
+ .equal(id, Datum::Int(7))
+ .is_err());
+ assert!(builder
+ .between(&score, Datum::String("z".into()), Datum::String("a".into()))
+ .is_err());
+ assert!(builder
+ .not_between(&score, Datum::String("z".into()),
Datum::String("a".into()))
+ .is_err());
+}
+
+#[test]
+fn nested_field_resolution_uses_components_and_canonical_names() {
+ let fields = vec![DataField::new(
+ 10,
+ "Profile.Root".into(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ 11,
+ "Score.Value".into(),
+ DataType::Int(IntType::new()),
+ )])),
+ )];
+ let sensitive = PredicateBuilder::new(&fields);
+ assert!(sensitive
+ .nested_field("profile.root", &["Score.Value"])
+ .is_err());
+ assert!(sensitive.nested_field("Profile.Root", &[]).is_err());
+ assert!(sensitive
+ .nested_field("Profile.Root", &["Score", "Value"])
+ .is_err());
+ assert!(sensitive
+ .nested_field("Profile.Root", &["Score.Value", "child"])
+ .is_err());
+ let insensitive = PredicateBuilder::new_with_case_sensitive(&fields,
false);
+ let reference = insensitive
+ .nested_field("profile.root", &["score.value"])
+ .unwrap();
+ assert_eq!(
+ insensitive.equal(&reference, Datum::Int(7)).unwrap(),
+ sensitive
+ .equal(
+ sensitive
+ .nested_field("Profile.Root", &["Score.Value"])
+ .unwrap(),
+ Datum::Int(7)
+ )
+ .unwrap()
+ );
+ let ambiguous = vec![DataField::new(
+ 10,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![
+ DataField::new(11, "Score".into(), DataType::Int(IntType::new())),
+ DataField::new(12, "score".into(), DataType::Int(IntType::new())),
+ ])),
+ )];
+ assert!(PredicateBuilder::new_with_case_sensitive(&ambiguous, false)
+ .nested_field("profile", &["score"])
+ .is_err());
+}
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 6fac24c6..15c0cdc9 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -20,7 +20,8 @@ mod blob_fallback;
use super::blob_resolver::BlobReadLimiter;
use super::data_file_reader::{
append_null_row_id_column, attach_row_id, expand_selected_row_ids,
- file_index_selection_to_local_ranges, insert_column_at,
DataFileReadTiming, DataFileReader,
+ file_index_selection_to_local_ranges, insert_column_at,
project_residual_output,
+ DataFileReadTiming, DataFileReader,
};
use super::read_limit::take_limited_batch;
use crate::arrow::format::blob::DEFAULT_BLOB_READ_PARALLELISM;
@@ -111,8 +112,8 @@ pub(crate) struct DataEvolutionReader {
schema_manager: SchemaManager,
table_schema_id: i64,
table_fields: Vec<DataField>,
- /// read_type with _ROW_ID filtered out, widened with predicate-only
- /// columns (appended strictly at the END) — used for file reads and the
+ /// read_type with _ROW_ID filtered out, widened with complete predicate
+ /// ROW inputs and extra top-level columns — used for file reads and the
/// column-merge plan.
wide_file_read_type: Vec<DataField>,
/// Position of _ROW_ID in the original read_type, if requested.
@@ -121,6 +122,7 @@ pub(crate) struct DataEvolutionReader {
/// _ROW_ID if requested). Batches are projected back to this after the
/// residual filter.
output_schema: Arc<arrow_schema::Schema>,
+ output_read_type: Vec<DataField>,
/// Arrow schema of wide batches at the _ROW_ID attach point: the original
/// read_type columns in caller order, then the extra predicate columns.
wide_output_schema: Arc<arrow_schema::Schema>,
@@ -165,10 +167,9 @@ impl DataEvolutionReader {
.collect();
let output_schema = crate::arrow::build_read_arrow_schema(&read_type)?;
- // Widen the file read set with predicate columns not already projected
- // so the residual filter can evaluate every leaf. Extras land strictly
- // at the END: `project_output` relies on that to project the
- // final batch back to `read_type` by prefix. Predicate leaf indices
+ // Widen predicate ROW inputs as well as extra top-level columns. The
+ // root positions remain stable; project_output restores the original
+ // nested fields after filtering. Predicate leaf indices
// point into the table schema, so `file_fields` = `table_fields`.
let file_predicates = (!predicates.is_empty()).then(|| FilePredicates {
predicates: predicates.clone(),
@@ -181,8 +182,10 @@ impl DataEvolutionReader {
// (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()..]);
+ let mut wide_read_type = wide_file_read_type.clone();
+ if let Some(index) = projected_row_id_index {
+ wide_read_type.insert(index, read_type[index].clone());
+ }
// 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.
@@ -207,6 +210,7 @@ impl DataEvolutionReader {
wide_file_read_type,
row_id_index,
output_schema,
+ output_read_type: read_type,
wide_output_schema,
predicates,
blob_as_descriptor,
@@ -759,38 +763,15 @@ impl DataEvolutionReader {
}
}
- /// Project a wide batch back to the caller's read_type.
- ///
- /// 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`, plus an unprojected `_ROW_ID` a
- /// residual needs — and `_ROW_ID` insertion at `row_id_index` keeps them
- /// trailing.
+ /// Restore the original top-level prefix and nested projection after the
+ /// residual has evaluated against complete ROW inputs.
fn project_output(&self, filtered: RecordBatch) ->
crate::Result<RecordBatch> {
- let final_width = self.output_schema.fields().len();
- if filtered.num_columns() == final_width
- && !filtered
- .schema()
- .fields()
- .iter()
- .any(|field| super::row_sidecar::is_blob_data(field))
- {
- return Ok(filtered);
- }
- let columns = filtered.columns()[..final_width].to_vec();
- let projected = if columns.is_empty() {
- RecordBatch::try_new_with_options(
- self.output_schema.clone(),
- columns,
-
&arrow_array::RecordBatchOptions::new().with_row_count(Some(filtered.num_rows())),
- )
- } else {
- RecordBatch::try_new(self.output_schema.clone(), columns)
- };
- projected.map_err(|e| Error::UnexpectedError {
- message: format!("Failed to project data-evolution batch to
read_type: {e}"),
- source: Some(Box::new(e)),
- })
+ project_residual_output(
+ filtered,
+ &self.wide_file_read_type,
+ &self.output_read_type,
+ self.output_schema.clone(),
+ )
}
async fn finish_wide_batch(
@@ -1345,6 +1326,7 @@ fn predicate_references_any_field(
table_fields: &[DataField],
) -> bool {
match predicate {
+ Predicate::NestedLeaf { root, .. } =>
field_names.contains(root.name()),
Predicate::Leaf { column, index, .. } => {
// Never a BLOB column; resolving its placeholder index would force
// every BLOB to be resolved before filtering.
@@ -3240,6 +3222,172 @@ mod tests {
use std::path::{Path, PathBuf};
use tempfile::tempdir;
+ #[tokio::test]
+ async fn
nested_predicates_restore_projection_after_raw_and_column_merge_reads() {
+ use crate::arrow::format::create_format_writer;
+ use crate::spec::{RowType, VarCharType};
+ use arrow_array::{StringArray, StructArray};
+ use arrow_buffer::NullBuffer;
+ use arrow_schema::DataType as ArrowDataType;
+
+ let score = DataField::new(1, "score".into(),
DataType::Int(IntType::new()));
+ let label = DataField::new(
+ 2,
+ "label".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ );
+ let profile = DataField::new(
+ 0,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![score.clone(), label.clone()])),
+ );
+ let other = DataField::new(3, "other".into(),
DataType::Int(IntType::new()));
+ let fields = vec![profile.clone(), other.clone()];
+ let schema =
crate::arrow::build_read_arrow_schema(std::slice::from_ref(&profile)).unwrap();
+ let ArrowDataType::Struct(children) = schema.field(0).data_type() else
{
+ unreachable!()
+ };
+ let profiles = StructArray::new(
+ children.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![7, 9, 7])),
+ Arc::new(StringArray::from(vec!["match", "other", "hidden"])),
+ ],
+ Some(NullBuffer::from(vec![true, true, false])),
+ );
+ let batch = RecordBatch::try_new(schema.clone(),
vec![Arc::new(profiles)]).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/de_nested_residual";
+ let bucket_path = format!("{table_path}/bucket-0");
+ let mut files = Vec::new();
+ for (name, batch, columns) in [
+ ("profile.parquet", batch, vec!["profile"]),
+ (
+ "other.parquet",
+ RecordBatch::try_new(
+ crate::arrow::build_read_arrow_schema(&[other]).unwrap(),
+ vec![Arc::new(Int32Array::from(vec![10, 20, 30]))],
+ )
+ .unwrap(),
+ vec!["other"],
+ ),
+ ] {
+ let output = file_io
+ .new_output(&format!("{bucket_path}/{name}"))
+ .unwrap();
+ let mut writer =
+ create_format_writer(&output, batch.schema(), "zstd", 1, None,
None, None)
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let mut file = data_file(name, 100, 3, 1, Some(columns));
+ file.file_size = writer.close().await.unwrap().file_size as i64;
+ files.push(file);
+ }
+ let builder = PredicateBuilder::new(&fields);
+ let score_ref = builder.nested_field("profile", &["score"]).unwrap();
+ let predicate = builder.equal(&score_ref, Datum::Int(7)).unwrap();
+ for merged in [false, true] {
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path.clone())
+ .with_total_buckets(1)
+ .with_data_files(if merged {
+ files.clone()
+ } else {
+ files[..1].to_vec()
+ })
+ .build()
+ .unwrap();
+ for children in [
+ vec![score.clone()],
+ vec![label.clone()],
+ vec![label.clone(), score.clone()],
+ ] {
+ for row_id_at in [None, Some(0), Some(1)] {
+ let mut projected = vec![DataField::new(
+ 0,
+ "profile".into(),
+ DataType::Row(RowType::new(children.clone())),
+ )];
+ if let Some(index) = row_id_at {
+ projected.insert(index,
crate::spec::row_id_data_field());
+ }
+ let reader = DataEvolutionReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io.clone(), table_path.into()),
+ 0,
+ fields.clone(),
+ projected.clone(),
+ vec![predicate.clone()],
+ false,
+ HashSet::new(),
+ HashSet::new(),
+ false,
+ None,
+ )
+ .unwrap();
+ let batches = reader
+ .read(std::slice::from_ref(&split))
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ for batch in batches {
+ assert_eq!(
+ batch.schema(),
+
crate::arrow::build_read_arrow_schema(&projected).unwrap()
+ );
+ let profile_at = usize::from(row_id_at == Some(0));
+ let values = batch
+ .column(profile_at)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(values.num_columns(), children.len());
+ for (index, child) in children.iter().enumerate() {
+ if child.name() == "score" {
+ assert_eq!(
+ values
+ .column(index)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 7
+ );
+ } else {
+ assert_eq!(
+ values
+ .column(index)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "match"
+ );
+ }
+ }
+ if let Some(index) = row_id_at {
+ assert_eq!(
+ batch
+ .column(index)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .value(0),
+ 100
+ );
+ }
+ }
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn
blob_view_payloads_use_each_upstream_file_io_even_for_the_same_uri() {
let uri = "memory:/referenced-payload";
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index b97f051a..742771b3 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -527,8 +527,9 @@ impl DataFileReader {
return Ok(futures::stream::empty().boxed());
}
- let mut read_type = self.read_type.clone();
- let output_field_count = read_type.len();
+ let output_read_type = self.read_type.clone();
+ let output_schema =
crate::arrow::build_read_arrow_schema(&output_read_type)?;
+ let mut read_type = output_read_type.clone();
let table_fields = self.table_fields.clone();
let predicates = self.predicates.clone();
let projects_row_id = read_type
@@ -600,6 +601,13 @@ impl DataFileReader {
row_filter_factory: None,
file_fields: table_fields.clone(),
});
+ if post_read_filter {
+ // Nested residuals need the complete ROW even when the caller only
+ // projects a sibling. Widen before reconciliation; restoring only
+ // the top-level column count would leak the extra nested children.
+ read_type =
+ crate::arrow::residual::widen_scan_fields(&read_type,
residual_predicates.as_ref());
+ }
// The first version of the engine hook is deliberately limited to a
// schema-identical raw read. Schema-evolution readers retain their
exact
// post-filter until expression adaptation is proven for that path.
@@ -861,16 +869,8 @@ impl DataFileReader {
residual,
&read_type,
)?;
- if read_type.len() > output_field_count {
-
filtered.project(&(0..output_field_count).collect::<Vec<_>>()).map_err(|e| {
- Error::UnexpectedError {
- message: format!("Failed to project filtered
RecordBatch: {e}"),
- source: Some(Box::new(e)),
- }
- })?
- } else {
- filtered
- }
+ let output_schema =
super::row_sidecar::propagate_blob_data(&output_schema, &filtered);
+ project_residual_output(filtered, &read_type,
&output_read_type, output_schema)?
} else {
result
};
@@ -1059,6 +1059,44 @@ impl DataFileReader {
}
}
+/// Restore an output projection after reading complete predicate ROW inputs.
+/// Columns retain their prefix order; ROW children are paired by Paimon field
+/// identity rather than Arrow position, including reordered nested
projections.
+pub(super) fn project_residual_output(
+ batch: RecordBatch,
+ source_fields: &[DataField],
+ output_fields: &[DataField],
+ output_schema: Arc<arrow_schema::Schema>,
+) -> crate::Result<RecordBatch> {
+ if batch.schema() == output_schema {
+ return Ok(batch);
+ }
+ let columns = output_fields
+ .iter()
+ .enumerate()
+ .map(|(index, target)| {
+ let source = source_fields
+ .iter()
+ .find(|field| field.id() == target.id())
+ .unwrap_or(target);
+ if source == target {
+ Ok(batch.column(index).clone())
+ } else {
+
crate::arrow::nested_evolution::evolve_field(batch.column(index), source,
target)
+ }
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ RecordBatch::try_new_with_options(
+ output_schema,
+ columns,
+
&arrow_array::RecordBatchOptions::new().with_row_count(Some(batch.num_rows())),
+ )
+ .map_err(|error| Error::UnexpectedError {
+ message: format!("Failed to restore residual output projection:
{error}"),
+ source: Some(Box::new(error)),
+ })
+}
+
/// Reconcile one decoded column with the read schema's type for it.
///
/// Delegates to [`crate::arrow::nested_evolution::evolve_column`], which
returns
@@ -2240,6 +2278,158 @@ mod row_tests {
assert_eq!(row_ids(&selected), vec![101]);
}
+ #[tokio::test]
+ async fn
nested_residual_reads_unprojected_sibling_before_restoring_output() {
+ use arrow_array::StructArray;
+ use arrow_buffer::NullBuffer;
+ use arrow_schema::DataType as ArrowDataType;
+
+ for (evolved, row_id) in [(true, false), (false, true), (true, true)] {
+ let score = field(1, "score", DataType::Int(IntType::new()));
+ let label = field(2, "label",
DataType::VarChar(VarCharType::string_type()));
+ let physical = vec![field(
+ 0,
+ "profile",
+ DataType::Row(RowType::new(vec![score.clone(),
label.clone()])),
+ )];
+ let mut current = physical.clone();
+ if evolved {
+ current[0] = field(
+ 0,
+ "profile",
+ DataType::Row(RowType::new(vec![
+ field(1, "score", DataType::BigInt(BigIntType::new())),
+ label.clone(),
+ ])),
+ );
+ }
+ let schema = build_target_arrow_schema(&physical).unwrap();
+ let ArrowDataType::Struct(children) = schema.field(0).data_type()
else {
+ unreachable!()
+ };
+ let values = StructArray::new(
+ children.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![7, 9, 7])),
+ Arc::new(StringArray::from(vec!["match", "other",
"hidden"])),
+ ],
+ Some(NullBuffer::from(vec![true, true, false])),
+ );
+ let batch = RecordBatch::try_new(schema.clone(),
vec![Arc::new(values)]).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path =
format!("memory:/nested_residual_{evolved}_{row_id}");
+ let bucket_path = format!("{table_path}/bucket-0");
+ let output = file_io
+ .new_output(&format!("{bucket_path}/part.parquet"))
+ .unwrap();
+ let mut writer = create_format_writer(&output, schema, "zstd", 1,
None, None, None)
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let file_size = writer.close().await.unwrap().file_size as i64;
+ let mut file_meta = data_file("part.parquet", file_size, 3, 1);
+ file_meta.first_row_id = Some(100);
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![file_meta.clone()])
+ .build()
+ .unwrap();
+ let mut projected = vec![field(
+ 0,
+ "profile",
+ DataType::Row(RowType::new(vec![label])),
+ )];
+ if row_id {
+ projected.insert(0, crate::spec::row_id_data_field());
+ }
+ let builder = PredicateBuilder::new(¤t);
+ for op in [
+ crate::spec::PredicateOperator::Eq,
+ crate::spec::PredicateOperator::IsNull,
+ ] {
+ let literals = if op == crate::spec::PredicateOperator::IsNull
{
+ vec![]
+ } else if evolved {
+ vec![Datum::Long(7)]
+ } else {
+ vec![Datum::Int(7)]
+ };
+ let score = builder.nested_field("profile",
&["score"]).unwrap();
+ let predicate = if op ==
crate::spec::PredicateOperator::IsNull {
+ builder.is_null(&score)
+ } else {
+ builder.equal(&score, literals[0].clone())
+ }
+ .unwrap();
+ let reader = DataFileReader::new(
+ file_io.clone(),
+ SchemaManager::new(file_io.clone(), table_path.clone()),
+ 2,
+ current.clone(),
+ projected.clone(),
+ vec![predicate],
+ );
+ let batches = reader
+ .read_single_file_stream(
+ &split,
+ file_meta.clone(),
+ evolved.then(|| physical.clone()),
+ None,
+ None,
+ )
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ for batch in batches {
+ assert_eq!(
+ batch.schema(),
+
crate::arrow::build_read_arrow_schema(&projected).unwrap()
+ );
+ let profile = batch
+ .column(usize::from(row_id))
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(profile.num_columns(), 1);
+ if op == crate::spec::PredicateOperator::IsNull {
+ assert!(profile.is_null(0));
+ } else {
+ assert_eq!(
+ profile
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "match"
+ );
+ }
+ if row_id {
+ assert_eq!(
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .value(0),
+ if op == crate::spec::PredicateOperator::IsNull {
+ 102
+ } else {
+ 100
+ }
+ );
+ }
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn read_row_id_with_data_predicate_keeps_physical_position() {
for format in ["row", "parquet"] {
diff --git a/crates/paimon/src/table/full_text_search_builder/filter.rs
b/crates/paimon/src/table/full_text_search_builder/filter.rs
index 37246cca..eef7a1a9 100644
--- a/crates/paimon/src/table/full_text_search_builder/filter.rs
+++ b/crates/paimon/src/table/full_text_search_builder/filter.rs
@@ -67,6 +67,9 @@ fn ranges(ids: &RoaringTreemap) ->
crate::Result<Vec<RowRange>> {
fn field_ids(predicate: &Predicate, fields: &[DataField], ids: &mut
HashSet<i32>) {
match predicate {
+ Predicate::NestedLeaf { root, .. } => {
+ ids.insert(root.id());
+ }
Predicate::Leaf { column, .. } => {
if let Some(id) = find_field_id_by_name(fields, column) {
ids.insert(id);
diff --git a/crates/paimon/src/table/global_index_scanner.rs
b/crates/paimon/src/table/global_index_scanner.rs
index 71dd09a0..397d3766 100644
--- a/crates/paimon/src/table/global_index_scanner.rs
+++ b/crates/paimon/src/table/global_index_scanner.rs
@@ -387,6 +387,7 @@ impl GlobalIndexScanner {
) -> bool {
use crate::spec::PredicateOperator;
match predicate {
+ Predicate::NestedLeaf { .. } => false,
Predicate::Leaf { column, op, .. } => {
predicates::is_sorted_global_index_supported_op(*op)
&& !matches!(
diff --git a/crates/paimon/src/table/global_index_scanner/evaluator.rs
b/crates/paimon/src/table/global_index_scanner/evaluator.rs
index cc3a6a37..8a4530d5 100644
--- a/crates/paimon/src/table/global_index_scanner/evaluator.rs
+++ b/crates/paimon/src/table/global_index_scanner/evaluator.rs
@@ -90,6 +90,7 @@ impl GlobalIndexScanner {
fields: &HashSet<i32>,
) -> Result<bool> {
match predicate {
+ Predicate::NestedLeaf { .. } => Ok(false),
Predicate::Leaf { column, .. } => Ok(self
.find_field_id_by_name(column)?
.is_some_and(|id| fields.contains(&id))),
@@ -814,6 +815,9 @@ impl GlobalIndexScanner {
field_ids: &mut HashSet<i32>,
) -> Result<()> {
match predicate {
+ Predicate::NestedLeaf { root, .. } => {
+ field_ids.insert(root.id());
+ }
Predicate::Leaf { column, .. } => {
if let Some(field_id) = self.find_field_id_by_name(column)? {
field_ids.insert(field_id);
diff --git a/crates/paimon/src/table/managed_blob_reader.rs
b/crates/paimon/src/table/managed_blob_reader.rs
index 71379474..02bb0a7a 100644
--- a/crates/paimon/src/table/managed_blob_reader.rs
+++ b/crates/paimon/src/table/managed_blob_reader.rs
@@ -249,10 +249,17 @@ impl ManagedBlobReadPlan {
message: format!("Managed BLOB output projection
is missing a column: {error}"),
source: Some(Box::new(error)),
})?;
- yield batch.project(&indices).map_err(|error|
crate::Error::DataInvalid {
+ let output = batch.project(&indices).map_err(|error|
crate::Error::DataInvalid {
message: format!("Failed to project managed BLOB read
output: {error}"),
source: Some(Box::new(error)),
})?;
+ let source_fields = indices.iter()
+ .map(|index|
self.scan_fields[*index].clone()).collect::<Vec<_>>();
+ let schema =
crate::arrow::build_read_arrow_schema(&self.output_fields)?;
+ let schema =
super::row_sidecar::propagate_blob_data(&schema, &output);
+ yield super::data_file_reader::project_residual_output(
+ output, &source_fields, &self.output_fields, schema,
+ )?;
}
}
})
@@ -331,7 +338,7 @@ impl<'a> BlobPredicateBatch<'a> {
.map(|(active, matched)| *active && !matched)
.collect())
}
- Predicate::Leaf { op, .. } => {
+ Predicate::Leaf { op, .. } | Predicate::NestedLeaf { op, .. }
=> {
let fields = self
.blob_fields
.iter()
@@ -739,6 +746,100 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn nested_and_blob_predicates_restore_pruned_row_output() {
+ use crate::spec::{
+ BlobType, DataType, Datum, IntType, PredicateBuilder, RowType,
VarCharType,
+ };
+ use arrow_array::{Int32Array, StringArray};
+ use futures::TryStreamExt;
+
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ io.new_output("memory:/nested-blob")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"yes"))
+ .await
+ .unwrap();
+ let descriptor = BlobDescriptor::new("memory:/nested-blob".into(), 0,
3).serialize();
+ let score = DataField::new(10, "score".into(),
DataType::Int(IntType::new()));
+ let label = DataField::new(
+ 11,
+ "label".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ );
+ let fields = vec![
+ DataField::new(
+ 0,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![score, label.clone()])),
+ ),
+ DataField::new(1, "payload".into(),
DataType::Blob(BlobType::new())),
+ ];
+ let output = vec![DataField::new(
+ 0,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![label])),
+ )];
+ let schema = crate::arrow::build_read_arrow_schema(&fields).unwrap();
+ let ArrowDataType::Struct(children) = schema.field(0).data_type() else
{
+ unreachable!()
+ };
+ let profiles = StructArray::new(
+ children.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![7, 8])),
+ Arc::new(StringArray::from(vec!["selected", "discarded"])),
+ ],
+ None,
+ );
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(profiles),
+ Arc::new(LargeBinaryArray::from(vec![
+ descriptor.as_slice(),
+ descriptor.as_slice(),
+ ])),
+ ],
+ )
+ .unwrap();
+ let builder = PredicateBuilder::new(&fields);
+ let score = builder.nested_field("profile", &["score"]).unwrap();
+ let predicate = Predicate::and(vec![
+ builder.equal(&score, Datum::Int(7)).unwrap(),
+ builder
+ .equal("payload", Datum::Bytes(b"yes".to_vec()))
+ .unwrap(),
+ ]);
+ let values = HashMap::new();
+ let options = CoreOptions::new(&values);
+ let plan = ManagedBlobReadPlan::new(&output, &[predicate], &fields,
&options).unwrap();
+ let stream = Box::pin(futures::stream::iter(vec![Ok(batch)]));
+ let batches: Vec<_> = plan
+ .finish(stream, &options, io, 1, None)
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let batch = &batches[0];
+ assert_eq!(batch.num_rows(), 1);
+ assert_eq!(
+ batch.schema(),
+ crate::arrow::build_read_arrow_schema(&output).unwrap()
+ );
+ let profile = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let label = profile
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(label.value(0), "selected");
+ }
+
#[tokio::test]
async fn
predicate_and_output_share_payload_bytes_resembling_a_descriptor() {
use crate::spec::{BlobType, DataType, Datum, PredicateBuilder};
diff --git a/crates/paimon/src/table/partition_filter.rs
b/crates/paimon/src/table/partition_filter.rs
index ce5dda04..07ee5712 100644
--- a/crates/paimon/src/table/partition_filter.rs
+++ b/crates/paimon/src/table/partition_filter.rs
@@ -213,6 +213,7 @@ fn predicate_may_match(
partition_fields: &[DataField],
) -> bool {
match predicate {
+ Predicate::NestedLeaf { .. } => true,
Predicate::AlwaysTrue => true,
Predicate::AlwaysFalse => false,
Predicate::And(children) => children
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
index eb853077..5f409cc9 100644
--- a/crates/paimon/src/table/query_auth.rs
+++ b/crates/paimon/src/table/query_auth.rs
@@ -183,6 +183,9 @@ pub(crate) fn leaf_names(
fn collect(predicate: &crate::spec::Predicate, out: &mut
std::collections::HashSet<String>) {
use crate::spec::Predicate;
match predicate {
+ Predicate::NestedLeaf { root, .. } => {
+ out.insert(root.name().to_string());
+ }
Predicate::Leaf { column, .. } => {
out.insert(column.clone());
}
@@ -290,6 +293,7 @@ pub(crate) fn reject_noncanonical_leaves(
use crate::spec::Predicate;
fn check(predicate: &Predicate, fields: &[crate::spec::DataField]) ->
crate::Result<()> {
match predicate {
+ Predicate::NestedLeaf { .. } =>
predicate.validate_nested_fields(fields),
Predicate::Leaf {
column,
index,
diff --git a/crates/paimon/src/table/query_auth/rules.rs
b/crates/paimon/src/table/query_auth/rules.rs
index 83387330..cd9dd9dc 100644
--- a/crates/paimon/src/table/query_auth/rules.rs
+++ b/crates/paimon/src/table/query_auth/rules.rs
@@ -192,6 +192,13 @@ pub(super) fn rule_mask(
batch_fields: &[DataField],
) -> Result<BooleanArray> {
match predicate {
+ Predicate::NestedLeaf { .. } =>
crate::arrow::residual::evaluate_predicate_mask(
+ batch,
+ predicate,
+ schema_fields,
+ batch_fields,
+ )?
+ .ok_or_else(|| unsupported("nested row filter cannot be evaluated")),
Predicate::AlwaysTrue => Ok(BooleanArray::from(vec![true;
batch.num_rows()])),
Predicate::AlwaysFalse => Ok(BooleanArray::from(vec![false;
batch.num_rows()])),
Predicate::And(children) | Predicate::Or(children) => {
diff --git a/crates/paimon/src/table/row_id_predicate.rs
b/crates/paimon/src/table/row_id_predicate.rs
index b6253fef..48361d61 100644
--- a/crates/paimon/src/table/row_id_predicate.rs
+++ b/crates/paimon/src/table/row_id_predicate.rs
@@ -116,6 +116,7 @@ pub(crate) fn reject_row_id_filter(predicates:
&[Predicate], read: &str) -> crat
/// Whether any leaf of `predicate` references `_ROW_ID`.
pub(crate) fn references_row_id(predicate: &Predicate) -> bool {
match predicate {
+ Predicate::NestedLeaf { .. } => false,
Predicate::Leaf { column, .. } => is_row_id_column(column),
Predicate::And(children) | Predicate::Or(children) => {
children.iter().any(references_row_id)
diff --git a/crates/paimon/src/table/stats_filter.rs
b/crates/paimon/src/table/stats_filter.rs
index accc320f..fb956443 100644
--- a/crates/paimon/src/table/stats_filter.rs
+++ b/crates/paimon/src/table/stats_filter.rs
@@ -575,6 +575,7 @@ fn data_evolution_predicate_may_match(
row_count: i64,
) -> bool {
match predicate {
+ Predicate::NestedLeaf { .. } => true,
Predicate::AlwaysTrue => true,
Predicate::AlwaysFalse => false,
Predicate::And(children) => children.iter().all(|child| {
@@ -640,6 +641,7 @@ fn data_evolution_predicate_must_match(
row_count: i64,
) -> bool {
match predicate {
+ Predicate::NestedLeaf { .. } => false,
Predicate::AlwaysTrue => row_count > 0,
Predicate::AlwaysFalse => false,
Predicate::And(children) => children.iter().all(|child| {
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 4ae7b711..ed8047f9 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -1295,7 +1295,7 @@ impl<'a> PaimonTableRead<'a> {
if let Some(options) = blob_options {
return self.finish_authorized_blobs(stream, &batch_fields,
&pending, options);
}
- let stream = project_authorized_stream(stream, self.read_type.len());
+ let stream = project_authorized_stream(stream, &batch_fields,
&self.read_type)?;
Ok(match self.limit {
None => Box::pin(stream),
Some(limit) => limit_rows(Box::pin(stream), limit),
@@ -1409,7 +1409,23 @@ impl<'a> PaimonTableRead<'a> {
// The payload filter owns its widened schema. Drop
authorization
// columns before it resolves columns by position.
let fields = plan.scan_fields().to_vec();
- Box::pin(stream.map(move |batch| {
+ let sources = fields
+ .iter()
+ .map(|field| {
+ batch_fields
+ .iter()
+ .find(|source| source.id() == field.id())
+ .cloned()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "authorized BLOB filter is missing field
'{}'",
+ field.name()
+ ),
+ source: None,
+ })
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
+ let stream = Box::pin(stream.map(move |batch| {
let batch = batch?;
let indices = fields
.iter()
@@ -1425,7 +1441,11 @@ impl<'a> PaimonTableRead<'a> {
message: format!("failed to project the authorized
batch: {e}"),
source: Some(Box::new(e)),
})
- })) as ArrowRecordBatchStream
+ })) as ArrowRecordBatchStream;
+ // An ordinary predicate may have read extra ROW children that
+ // the remaining BLOB predicates do not need. Restore the
plan's
+ // nested inputs using their actual, pre-projection types.
+ project_authorized_stream(stream, &sources,
plan.scan_fields())?
}
None => {
let predicates = crate::arrow::format::FilePredicates {
@@ -1441,7 +1461,8 @@ impl<'a> PaimonTableRead<'a> {
&fields,
)
});
- let stream = project_authorized_stream(Box::pin(stream),
self.read_type.len());
+ let stream =
+ project_authorized_stream(Box::pin(stream), batch_fields,
&self.read_type)?;
let stream = match self.limit {
Some(limit) => limit_rows(stream, limit),
None => stream,
@@ -1715,17 +1736,24 @@ impl<'a> PaimonTableRead<'a> {
fn project_authorized_stream(
stream: ArrowRecordBatchStream,
- columns: usize,
-) -> ArrowRecordBatchStream {
- let projection: Vec<usize> = (0..columns).collect();
- Box::pin(stream.map(move |batch| {
- batch?
- .project(&projection)
- .map_err(|e| crate::Error::DataInvalid {
- message: format!("failed to project the authorized batch:
{e}"),
- source: Some(Box::new(e)),
- })
- }))
+ source_fields: &[DataField],
+ output_fields: &[DataField],
+) -> crate::Result<ArrowRecordBatchStream> {
+ let source_fields = source_fields.to_vec();
+ let output_fields = output_fields.to_vec();
+ let output_schema = crate::arrow::build_read_arrow_schema(&output_fields)?;
+ Ok(Box::pin(stream.map(move |batch| {
+ let batch = batch?;
+ let schema = super::row_sidecar::propagate_blob_data(&output_schema,
&batch);
+ // Predicates may widen a partially projected ROW to include a sibling.
+ // Restore its children as well as dropping authorization-only columns.
+ super::data_file_reader::project_residual_output(
+ batch,
+ &source_fields,
+ &output_fields,
+ schema,
+ )
+ })))
}
/// Cuts batches to at most `quota` rows; stages up to `spend_quota` must not
@@ -2983,6 +3011,326 @@ mod tests {
)
}
+ #[tokio::test]
+ async fn
test_authorized_nested_predicate_restores_projected_row_children() {
+ use crate::arrow::format::create_format_writer;
+ use crate::spec::{BlobDescriptor, BlobType, RowType, VarCharType};
+ use arrow_array::{Int32Array, LargeBinaryArray};
+
+ for (mode, blob, payload_filter, project_payload, mask_id) in [
+ ("ordinary", false, false, false, false),
+ ("descriptor", true, false, true, false),
+ ("payload", true, true, true, false),
+ ("hidden_payload", true, true, false, false),
+ ("masked_ordinary", false, false, false, true),
+ ("masked_descriptor", true, false, true, true),
+ ("masked_payload", true, true, true, true),
+ ] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column(
+ "profile",
+ DataType::Row(RowType::new(vec![
+ DataField::new(2, "score".into(),
DataType::Int(IntType::new())),
+ DataField::new(
+ 3,
+ "label".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ ])),
+ )
+ .option("query-auth.enabled", "true")
+ .option(
+ "blob-as-descriptor",
+ if payload_filter { "false" } else { "true" },
+ );
+ if blob {
+ schema = schema
+ .column("payload", DataType::Blob(BlobType::new()))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ let schema = schema.build().unwrap();
+ let table_schema = TableSchema::new(0, &schema);
+ let fields = table_schema.fields().to_vec();
+ let DataType::Row(profile_type) = fields[1].data_type() else {
+ unreachable!()
+ };
+ let profile_arrow =
crate::arrow::paimon_type_to_arrow(fields[1].data_type()).unwrap();
+ let arrow_schema::DataType::Struct(children) = profile_arrow else {
+ unreachable!()
+ };
+ let profiles = StructArray::new(
+ children,
+ vec![
+ Arc::new(Int32Array::from(vec![7, 9, 7])),
+ Arc::new(StringArray::from(vec!["selected", "other",
"denied"])),
+ ],
+ None,
+ );
+ let descriptor = BlobDescriptor::new("memory:/payload".into(), 0,
3).serialize();
+ let mut columns: Vec<ArrayRef> = vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3])),
+ Arc::new(profiles),
+ ];
+ if blob {
+ columns.push(Arc::new(LargeBinaryArray::from(vec![
+ descriptor.as_slice();
+ 3
+ ])));
+ }
+ let batch = RecordBatch::try_new(
+ crate::arrow::build_read_arrow_schema(&fields).unwrap(),
+ columns,
+ )
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ file_io
+ .new_output("memory:/payload")
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"abc"))
+ .await
+ .unwrap();
+ let location = format!("memory:/authorized_nested_{mode}");
+ let bucket_path = format!("{location}/bucket-0");
+ let output = file_io
+ .new_output(&format!("{bucket_path}/data.parquet"))
+ .unwrap();
+ let mut writer =
+ create_format_writer(&output, batch.schema(), "zstd", 1, None,
None, None)
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let mut data_file = file("data.parquet", 0, Some(0));
+ data_file.row_count = 3;
+ data_file.first_row_id = Some(100);
+ data_file.file_size = writer.close().await.unwrap().file_size as
i64;
+ let table = Table {
+ schema: table_schema,
+ schema_manager:
crate::table::schema_manager::SchemaManager::new(
+ file_io.clone(),
+ location.clone(),
+ ),
+ file_io,
+ location,
+ ..crate::table::rest_query_auth_table().await
+ };
+ let builder = PredicateBuilder::new(&fields);
+ let score = builder.nested_field("profile", &["score"]).unwrap();
+ let predicate = builder.equal(&score, Datum::Int(7)).unwrap();
+ let mut predicates = vec![predicate];
+ if payload_filter {
+ predicates.push(
+ builder
+ .equal("payload", Datum::Bytes(b"abc".to_vec()))
+ .unwrap(),
+ );
+ }
+ let grant = crate::table::query_auth::QueryAuthGrant::new(
+ table.query_auth_session().unwrap(),
+ None,
+ crate::table::query_auth::Rules {
+ filters: vec![builder.less_than("id",
Datum::Int(3)).unwrap().into()],
+ masks: if mask_id {
+ crate::table::query_auth::Rules::parse(
+ &crate::api::AuthTableQueryResponse {
+ filter: None,
+ column_masking:
Some(std::collections::HashMap::from([(
+ "id".to_string(),
+ r#"{"name":"NULL"}"#.to_string(),
+ )])),
+ },
+ &fields,
+ )
+ .unwrap()
+ .masks
+ } else {
+ Vec::new()
+ },
+ },
+ );
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_total_buckets(1)
+ .with_data_files(vec![data_file])
+ .build()
+ .unwrap()
+ .planned(Some(Arc::new(grant)));
+ let mut projected = vec![DataField::new(
+ fields[1].id(),
+ "profile".into(),
+
DataType::Row(RowType::new(vec![profile_type.fields()[1].clone()])),
+ )];
+ if mask_id {
+ projected.push(fields[0].clone());
+ }
+ if project_payload {
+ projected.push(fields[2].clone());
+ }
+ for row_kind in [false, true] {
+ let read = TableRead::new(&table, projected.clone(),
predicates.clone())
+ .with_limit(Some(1));
+ let stream = if row_kind {
+ read.to_arrow_with_row_kind(std::slice::from_ref(&split))
+ } else {
+ read.to_arrow(std::slice::from_ref(&split))
+ }
+ .unwrap();
+ let batches: Vec<_> = stream.try_collect().await.unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let mut expected = projected.clone();
+ if row_kind {
+ expected.insert(0, row_kind_field());
+ }
+ for batch in batches {
+ assert_eq!(
+ batch.schema(),
+
crate::arrow::build_read_arrow_schema(&expected).unwrap()
+ );
+ let profile = batch
+ .column(usize::from(row_kind))
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(profile.num_columns(), 1);
+ let label = profile
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(label.value(0), "selected");
+ if row_kind {
+ assert_eq!(
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "+I"
+ );
+ }
+ if project_payload {
+ let payload = batch
+ .column(usize::from(row_kind) + 1 +
usize::from(mask_id))
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap();
+ assert_eq!(
+ payload.value(0),
+ if payload_filter {
+ b"abc".as_slice()
+ } else {
+ descriptor.as_slice()
+ },
+ );
+ }
+ if mask_id {
+ let id = batch.column(usize::from(row_kind) + 1);
+ assert_eq!(id.null_count(), batch.num_rows());
+ assert!(batch
+ .schema()
+ .field(usize::from(row_kind) + 1)
+ .is_nullable());
+ }
+ }
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn
test_authorized_projection_preserves_null_parents_and_blob_markers() {
+ use crate::spec::{BlobType, RowType, VarCharType};
+ use arrow_array::{Int32Array, LargeBinaryArray};
+ use arrow_buffer::NullBuffer;
+
+ let score = DataField::new(11, "score".into(),
DataType::Int(IntType::new()));
+ let label = DataField::new(
+ 12,
+ "label".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ );
+ let hidden = DataField::new(13, "hidden".into(),
DataType::Int(IntType::new()));
+ let source = vec![
+ DataField::new(
+ 10,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![score.clone(), label.clone(),
hidden])),
+ ),
+ DataField::new(20, "payload".into(),
DataType::Blob(BlobType::new())),
+ DataField::new(30, "auth_only".into(),
DataType::Int(IntType::new())),
+ ];
+ let output = vec![
+ DataField::new(
+ 10,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![label, score])),
+ ),
+ source[1].clone(),
+ ];
+ let schema = crate::arrow::build_read_arrow_schema(&source).unwrap();
+ let arrow_schema::DataType::Struct(children) =
schema.field(0).data_type() else {
+ unreachable!()
+ };
+ let profile = StructArray::new(
+ children.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![7, 99])),
+ Arc::new(StringArray::from(vec!["visible",
"under-null-parent"])),
+ Arc::new(Int32Array::from(vec![100, 200])),
+ ],
+ Some(NullBuffer::from(vec![true, false])),
+ );
+ let schema = super::super::row_sidecar::mark_blob_data(
+ &schema,
+ &std::collections::HashSet::from(["payload".to_string()]),
+ );
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(profile),
+ Arc::new(LargeBinaryArray::from(vec![b"raw".as_slice(); 2])),
+ Arc::new(Int32Array::from(vec![1, 2])),
+ ],
+ )
+ .unwrap();
+ let stream = Box::pin(stream::iter([Ok(batch)]));
+ let batches: Vec<_> = project_authorized_stream(stream, &source,
&output)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let batch = &batches[0];
+ assert_eq!(batch.num_columns(), 2);
+ assert!(super::super::row_sidecar::is_blob_data(
+ batch.schema().field(1)
+ ));
+ let profile = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(profile.column_names(), ["label", "score"]);
+ assert!(profile.is_null(1));
+ let labels = profile
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(labels.value(0), "visible");
+ assert!(labels.is_null(1));
+ let scores = profile
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(scores.value(0), 7);
+ assert!(scores.is_null(1));
+ }
+
#[tokio::test]
async fn
test_a_restricted_read_refuses_a_variant_extraction_that_can_fail() {
use crate::spec::{
diff --git a/crates/paimon/src/table/table_update_delete.rs
b/crates/paimon/src/table/table_update_delete.rs
index 84b3af40..d61657f1 100644
--- a/crates/paimon/src/table/table_update_delete.rs
+++ b/crates/paimon/src/table/table_update_delete.rs
@@ -86,6 +86,7 @@ fn validate_predicate(
fields: &[crate::spec::DataField],
) -> crate::Result<()> {
match predicate {
+ Predicate::NestedLeaf { .. } =>
predicate.validate_nested_fields(fields)?,
Predicate::Leaf {
column,
index,
