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 42700442 Support explicit partition selectors in native table scans
(#1129)
42700442 is described below
commit 42700442d572a8c995b81fa84bf4066d15b51998
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 15 13:41:33 2026 +0800
Support explicit partition selectors in native table scans (#1129)
---
.../python/python/pypaimon_rust/datafusion.pyi | 7 +
bindings/python/src/read.rs | 30 ++
bindings/python/tests/test_read.py | 72 ++++
crates/paimon/src/table/read_builder.rs | 89 ++++-
crates/paimon/src/table/table_read.rs | 70 +++-
crates/paimon/tests/partition_filter_test.rs | 382 +++++++++++++++++++++
crates/paimon/tests/rest_catalog_test.rs | 111 ++++++
7 files changed, 750 insertions(+), 11 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 3b80fb45..8fdf00e5 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -144,6 +144,13 @@ class ReadBuilder:
"""Set the maximum number of concurrent BLOB range reads. Must be
positive."""
...
def with_filter(self, predicate: dict) -> "ReadBuilder": ...
+ def with_partition_filter(self, predicate: Optional[dict]) ->
"ReadBuilder":
+ """Replace scan-only partition selection; None restores derived
pruning.
+
+ Only partition fields are allowed. Ordinary filters still apply to read
+ rows. Query authorization rejects selectors on masked partition fields.
+ """
+ ...
def with_row_ranges(self, ranges: Sequence[tuple[int, int]]) ->
"ReadBuilder":
"""Set Data Evolution row ranges. Empty selects no rows; format tables
are unsupported."""
...
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 0cec6518..fc2ae2fc 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -127,6 +127,7 @@ pub struct PyReadBuilder {
read_type: Option<Vec<DataField>>,
limit: Option<usize>,
filter: Option<Predicate>,
+ partition_filter: Option<Predicate>,
row_ranges: Option<Vec<RowRange>>,
case_sensitive: bool,
blob_parallelism: Option<usize>,
@@ -141,6 +142,7 @@ impl PyReadBuilder {
read_type: None,
limit: None,
filter: None,
+ partition_filter: None,
row_ranges: None,
case_sensitive: true,
blob_parallelism: None,
@@ -189,6 +191,7 @@ impl PyReadBuilder {
read_type: None,
limit: None,
filter: None,
+ partition_filter: None,
row_ranges: None,
case_sensitive: true,
blob_parallelism: None,
@@ -291,6 +294,26 @@ impl PyReadBuilder {
Ok(slf)
}
+ /// Set a scan-only partition selector using table fields. None clears it.
+ #[pyo3(signature = (predicate))]
+ fn with_partition_filter<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ predicate: Option<&Bound<'_, PyDict>>,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ let filter = predicate
+ .map(|predicate| {
+ dict_to_table_predicate(predicate, slf.table.schema(),
slf.case_sensitive)
+ })
+ .transpose()?;
+ // Validation and all partition/query-auth semantics belong to core.
+ slf.table
+ .new_read_builder()
+ .with_partition_filter(filter.clone())
+ .map_err(to_py_err)?;
+ slf.partition_filter = filter;
+ Ok(slf)
+ }
+
/// Set inclusive row ID ranges for Data Evolution scan planning.
/// Planned splits carry the ranges used by readers.
fn with_row_ranges(
@@ -317,6 +340,7 @@ impl PyReadBuilder {
read_type: self.read_type.clone(),
limit: self.limit,
filter: self.filter.clone(),
+ partition_filter: self.partition_filter.clone(),
row_ranges: self.row_ranges.clone(),
case_sensitive: self.case_sensitive,
incremental_scan: None,
@@ -379,6 +403,7 @@ pub struct PyTableScan {
read_type: Option<Vec<DataField>>,
limit: Option<usize>,
filter: Option<Predicate>,
+ partition_filter: Option<Predicate>,
row_ranges: Option<Vec<RowRange>>,
case_sensitive: bool,
incremental_scan: Option<PyIncrementalScan>,
@@ -479,6 +504,11 @@ impl PyTableScan {
&self.filter,
self.case_sensitive,
)?;
+ if let Some(filter) = &self.partition_filter {
+ builder
+ .with_partition_filter(Some(filter.clone()))
+ .map_err(to_py_err)?;
+ }
if let Some(row_ranges) = &self.row_ranges {
builder.with_row_ranges(row_ranges.clone());
}
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index b98e5a59..4a47525d 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -1814,3 +1814,75 @@ def
test_numeric_timestamp_window_options_use_core_scan(tmp_path, window, snapsh
batches = list(builder.new_read().read(plan.splits()))
actual = [row for batch in batches for row in batch.to_pylist()]
assert sorted(actual, key=lambda row: row["id"]) == expected
+
+
+def _read_partition_ids(builder, plan):
+ return sorted(row["id"] for batch in
builder.new_read().read_arrow(plan.splits())
+ for row in batch.to_pylist())
+
+
[email protected]("selector_first", [False, True])
+def
test_explicit_partition_selector_is_independent_from_the_row_filter(selector_first):
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_partitioned_table(warehouse)
+ builder = table.new_read_builder().with_projection(["id"])
+ selector = {"method": "in", "field": "dt", "literals": ["p1", "p2"]}
+ row_filter = {"method": "equal", "field": "dt", "literals": ["p2"]}
+ if selector_first:
+ builder.with_partition_filter(selector).with_filter(row_filter)
+ else:
+ builder.with_filter(row_filter).with_partition_filter(selector)
+ assert _read_partition_ids(builder, builder.new_scan().plan()) == [3]
+ # A scan-only constraint must not be applied as a reader residual.
+ selector_only =
table.new_read_builder().with_projection(["id"]).with_partition_filter(selector)
+ all_rows = table.new_read_builder().new_scan().plan()
+ assert _read_partition_ids(selector_only, all_rows) == [1, 2, 3, 4]
+ assert _read_partition_ids(selector_only,
selector_only.new_scan().plan()) == [1, 2, 3]
+ # The normal partition predicate must also apply with independent
splits.
+ normal =
table.new_read_builder().with_projection(["id"]).with_filter(row_filter)
+ assert _read_partition_ids(normal, all_rows) == [3]
+
+
+def
test_partition_selector_replacement_clear_validation_and_case_sensitivity():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_partitioned_table(warehouse)
+ builder =
table.new_read_builder().with_projection(["id"]).with_case_sensitive(False)
+ def selector(value):
+ return {"method": "equal", "field": "DT", "literals": [value]}
+ builder.with_partition_filter(selector("p1"))
+ scan = builder.new_scan()
+ builder.with_partition_filter(selector("p2"))
+ assert _read_partition_ids(builder, scan.plan()) == [1, 2]
+ assert _read_partition_ids(builder, builder.new_scan().plan()) == [3]
+ with pytest.raises(ValueError, match="only partition keys"):
+ builder.with_partition_filter({"method": "or", "children": [
+ selector("p3"), {"method": "equal", "field": "id", "literals":
[4]}]})
+ assert _read_partition_ids(builder, builder.new_scan().plan()) == [3]
+ builder.with_partition_filter(None)
+ assert _read_partition_ids(builder, builder.new_scan().plan()) == [1,
2, 3, 4]
+ with pytest.raises(ValueError):
+ table.new_read_builder().with_partition_filter(selector("p1"))
+
+
+def test_partition_selector_survives_incremental_and_stream_scan_capture():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_partitioned_table(warehouse)
+ builder =
table.new_read_builder().with_projection(["id"]).with_partition_filter(
+ {"method": "equal", "field": "dt", "literals": ["p1"]})
+ incremental = builder.new_incremental_scan(0, 1)
+ stream = builder.new_stream_scan()
+ builder.with_partition_filter({"method": "equal", "field": "dt",
"literals": ["p3"]})
+ assert _read_partition_ids(builder, incremental.plan()) == [1, 2]
+ assert _read_partition_ids(builder, stream.plan()) == [1, 2]
+ assert stream.checkpoint() == 2
+ stream.restore(1)
+ assert _read_partition_ids(builder, stream.plan()) == [1, 2]
+
+
+def
test_conflicting_explicit_selector_and_ordinary_partition_filter_return_no_rows():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_partitioned_table(warehouse)
+ builder =
table.new_read_builder().with_projection(["id"]).with_limit(1)
+ builder.with_partition_filter({"method": "equal", "field": "dt",
"literals": ["p1"]})
+ builder.with_filter({"method": "equal", "field": "dt", "literals":
["p3"]})
+ assert _read_partition_ids(builder, builder.new_scan().plan()) == []
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index 42ee2f3b..756764c3 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -222,6 +222,40 @@ impl<'a> ReadBuilder<'a> {
self
}
+ /// Select partitions at scan time, independently of the row filter.
+ ///
+ /// The predicate uses table-field indices and may reference only partition
+ /// keys. Like Java's `ReadBuilder.withPartitionFilter`, this replaces the
+ /// partition pruning derived from `with_filter`; the row filter still
applies
+ /// during reading. `None` restores derived partition pruning. Query-auth
+ /// scans reject selectors on masked keys because they cannot be checked
+ /// against the masked values after scanning. Format tables are
unsupported.
+ pub fn with_partition_filter(&mut self, filter: Option<Predicate>) ->
Result<&mut Self> {
+ let table = match &self.kind {
+ ReadBuilderKind::Paimon(builder) => builder.table,
+ ReadBuilderKind::Format(_) => {
+ return Err(Error::Unsupported {
+ message: "Explicit partition filters are not supported for
format tables"
+ .into(),
+ })
+ }
+ };
+ if let Some(filter) = &filter {
+ super::partition_filter::validate_partition_filter(table, filter)?;
+ if table.schema().core_options().query_auth_enabled() {
+ super::query_auth::reject_noncanonical_leaves(
+ std::slice::from_ref(filter),
+ table.schema().fields(),
+ )?;
+ }
+ }
+ match &mut self.kind {
+ ReadBuilderKind::Paimon(builder) => builder.partition_filter =
filter.map(Box::new),
+ ReadBuilderKind::Format(_) => unreachable!("format tables rejected
above"),
+ }
+ Ok(self)
+ }
+
/// Whether a translated predicate is exact at the table-provider boundary.
pub fn is_exact_filter_pushdown(&self, filter: &Predicate) -> bool {
match &self.kind {
@@ -355,6 +389,7 @@ struct PaimonReadBuilder<'a> {
/// ("last projection setter wins").
projection_names: Option<Vec<String>>,
filter: NormalizedFilter,
+ partition_filter: Option<Box<Predicate>>,
limit: Option<usize>,
/// Ranges the caller set. Each `with_row_ranges` replaces them.
explicit_row_ranges: Option<Vec<RowRange>>,
@@ -373,6 +408,7 @@ impl<'a> PaimonReadBuilder<'a> {
read_type: None,
projection_names: None,
filter: NormalizedFilter::default(),
+ partition_filter: None,
limit: None,
explicit_row_ranges: None,
derived_row_ranges: None,
@@ -534,15 +570,24 @@ impl<'a> PaimonReadBuilder<'a> {
/// does return `Result` and surfaces the error to the caller. This keeps
the
/// scan infallible without silently discarding an error that a read would
hit.
pub fn new_scan(&self) -> TableScan<'a> {
- let partition_filter =
self.filter.partition_predicate.clone().map(|pred| {
+ let partition_predicate = match &self.partition_filter {
+ Some(filter) => extract_predicate_for_keys(
+ filter,
+ self.table.schema().fields(),
+ self.table.schema().partition_keys(),
+ ),
+ None => self.filter.partition_predicate.clone(),
+ };
+ let partition_filter = partition_predicate.map(|pred| {
PartitionFilter::from_predicate(pred,
&self.table.schema().partition_fields())
});
let read_type = self.resolve_read_type().unwrap_or(None);
+ let data_predicates = self.scan_data_predicates();
let scan_predicates =
- super::managed_blob_reader::scan_predicates(self.table,
&self.filter.data_predicates);
+ super::managed_blob_reader::scan_predicates(self.table,
&data_predicates);
// A payload predicate is evaluated after descriptor resolution. A
// scan limit must not count rows which may fail that residual filter.
- let scan_limit = if scan_predicates.len() ==
self.filter.data_predicates.len() {
+ let scan_limit = if scan_predicates.len() == data_predicates.len() {
self.limit
} else {
None
@@ -557,7 +602,7 @@ impl<'a> PaimonReadBuilder<'a> {
)
.with_projected_read_field_ids(projected_read_field_ids_with_predicates(
&read_type,
- &self.filter.data_predicates,
+ &data_predicates,
self.table.schema().fields(),
))
}
@@ -590,11 +635,43 @@ impl<'a> PaimonReadBuilder<'a> {
None => configured_parquet_read_budget(self.table)?,
};
TableRead::new(self.table, read_type,
self.filter.data_predicates.clone())
+ .with_partition_predicate(self.filter.partition_predicate.clone())
.with_parquet_read_budget(parquet_read_budget)
.with_blob_parallelism(self.blob_parallelism)
.map(|read| read.with_limit(self.limit))
}
+ /// An explicit selector replaces derived partition pruning. Retain the
+ /// ordinary partition conjuncts for safe stats/LIMIT decisions, rebound
from
+ /// the partition row to the table layout (partition order may differ).
+ fn scan_data_predicates(&self) -> Vec<Predicate> {
+ let mut predicates = self.filter.data_predicates.clone();
+ if self.partition_filter.is_none() {
+ return predicates;
+ }
+ if let Some(partition) = &self.filter.partition_predicate {
+ let mapping = self
+ .table
+ .schema()
+ .partition_keys()
+ .iter()
+ .map(|name| {
+ self.table
+ .schema()
+ .fields()
+ .iter()
+ .position(|field| field.name() == name)
+ })
+ .collect::<Vec<_>>();
+ predicates.push(
+ partition
+ .remap_field_index(&mapping)
+ .expect("partition fields belong to the table schema"),
+ );
+ }
+ predicates
+ }
+
/// Resolve the effective read type, deferring projection name resolution
to
/// the case sensitivity effective at build time (order-independent with
/// `with_case_sensitive`).
@@ -2118,9 +2195,11 @@ mod tests {
None,
);
+ let mut partition = crate::spec::BinaryRowBuilder::new(1);
+ partition.write_string(0, "2024-01-01");
let split = DataSplitBuilder::new()
.with_snapshot(1)
- .with_partition(BinaryRow::new(1))
+ .with_partition(partition.build())
.with_bucket(0)
.with_bucket_path(local_file_path(&bucket_dir))
.with_total_buckets(1)
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 6242a905..598e8bc9 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -40,6 +40,7 @@ use arrow_schema::Schema as ArrowSchema;
use arrow_select::concat::concat as arrow_concat;
use arrow_select::take::take;
use futures::{stream, StreamExt};
+use std::borrow::Cow;
use std::cmp::Ordering;
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
use std::sync::Arc;
@@ -154,6 +155,18 @@ impl<'a> TableRead<'a> {
}
}
+ /// Ordinary partition conjuncts are evaluated against split metadata,
+ /// since partition columns may be absent from physical data files. The
+ /// predicate uses partition-row indices, not table-field indices.
+ pub(crate) fn with_partition_predicate(self, predicate: Option<Predicate>)
-> Self {
+ match self.0 {
+ TableReadKind::Paimon(read) => Self(TableReadKind::Paimon(
+ read.with_partition_predicate(predicate),
+ )),
+ TableReadKind::Format(_) => self,
+ }
+ }
+
/// Set the maximum number of concurrent BLOB range reads for this read.
pub fn with_blob_parallelism(self, blob_parallelism: usize) ->
crate::Result<Self> {
if blob_parallelism == 0 {
@@ -316,6 +329,7 @@ struct PaimonTableRead<'a> {
table: &'a Table,
read_type: Vec<DataField>,
data_predicates: Vec<Predicate>,
+ partition_predicate: Option<Predicate>,
row_filter_factory: Option<Arc<dyn crate::arrow::RowFilterFactory>>,
parquet_read_budget: Option<Arc<ReadBudget>>,
resources: Option<ResourceContext>,
@@ -335,6 +349,7 @@ impl<'a> PaimonTableRead<'a> {
table,
read_type,
data_predicates,
+ partition_predicate: None,
row_filter_factory: None,
parquet_read_budget: None,
resources: None,
@@ -365,7 +380,8 @@ impl<'a> PaimonTableRead<'a> {
/// [`ReadBuilder::with_filter`](crate::table::ReadBuilder::with_filter)
/// for per-format exceptions).
pub fn with_filter(mut self, filter: Predicate) -> Self {
- let (_, data_predicates) = split_scan_predicates(self.table, filter);
+ let (partition_predicate, data_predicates) =
split_scan_predicates(self.table, filter);
+ self.partition_predicate = partition_predicate;
// Keep the FULL data predicate (including `And`/`Or`/`Not`). Native
// pushdown / stats pruning skip compound nodes they cannot use, and
the
// residual pass applies the full predicate exactly. Pruning here would
@@ -374,6 +390,30 @@ impl<'a> PaimonTableRead<'a> {
self
}
+ fn with_partition_predicate(mut self, predicate: Option<Predicate>) ->
Self {
+ self.partition_predicate = predicate;
+ self
+ }
+
+ /// Clone split metadata only if at least one split is rejected.
Authorization
+ /// must be checked against the original list before calling this method.
+ fn matching_splits<'s>(&self, splits: &'s [DataSplit]) ->
crate::Result<Cow<'s, [DataSplit]>> {
+ let Some(predicate) = &self.partition_predicate else {
+ return Ok(Cow::Borrowed(splits));
+ };
+ let mut retained: Option<Vec<DataSplit>> = None;
+ for (index, split) in splits.iter().enumerate() {
+ if crate::spec::eval_row(predicate, split.partition())? {
+ if let Some(retained) = &mut retained {
+ retained.push(split.clone());
+ }
+ } else {
+ retained.get_or_insert_with(|| splits[..index].to_vec());
+ }
+ }
+ Ok(retained.map_or(Cow::Borrowed(splits), Cow::Owned))
+ }
+
fn with_row_filter_factory(mut self, factory: Arc<dyn
crate::arrow::RowFilterFactory>) -> Self {
self.row_filter_factory = Some(factory);
self
@@ -428,6 +468,7 @@ impl<'a> PaimonTableRead<'a> {
}
// Delta / Changelog rows are read as-is from planned files (no
full-table
// merge against historical base versions).
+ let data_splits = self.matching_splits(&data_splits)?;
self.new_data_file_reader()?.read(&data_splits)
}
@@ -440,6 +481,7 @@ impl<'a> PaimonTableRead<'a> {
let table = self.table.clone();
let read_type = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
+ let partition_predicate = self.partition_predicate.clone();
let parquet_read_budget = self.parquet_read_budget()?;
let blob_parallelism = self.blob_parallelism;
@@ -448,9 +490,11 @@ impl<'a> PaimonTableRead<'a> {
let table = table.clone();
let read_type = read_type.clone();
let data_predicates = data_predicates.clone();
+ let partition_predicate = partition_predicate.clone();
let parquet_read_budget = Arc::clone(&parquet_read_budget);
let worker: ArrowRecordBatchStream =
Box::pin(async_stream::try_stream! {
let pair_read = PaimonTableRead::new(&table, read_type,
data_predicates)
+ .with_partition_predicate(partition_predicate)
.with_parquet_read_budget(parquet_read_budget)
.with_blob_parallelism(blob_parallelism);
let mut pair_stream =
pair_read.to_diff_after_image_stream(&before, &after)?;
@@ -499,6 +543,7 @@ impl<'a> PaimonTableRead<'a> {
"a row-kind read cannot apply a row filter or column masking",
));
}
+ let data_splits = self.matching_splits(data_splits)?;
let schema = audit_schema_for_read_type(&self.read_type, false)?;
let (streaming, materialized): (Vec<_>, Vec<_>) = data_splits
.iter()
@@ -538,6 +583,7 @@ impl<'a> PaimonTableRead<'a> {
let table = self.table.clone();
let read_type = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
+ let partition_predicate = self.partition_predicate.clone();
let row_filter_factory = self.row_filter_factory.clone();
let parquet_read_budget = self.parquet_read_budget()?;
let resources = self.resources.clone();
@@ -552,6 +598,7 @@ impl<'a> PaimonTableRead<'a> {
table: &table,
read_type: read_type.clone(),
data_predicates: data_predicates.clone(),
+ partition_predicate: partition_predicate.clone(),
row_filter_factory: row_filter_factory.clone(),
parquet_read_budget: Some(parquet_read_budget.clone()),
resources: resources.clone(),
@@ -590,6 +637,7 @@ impl<'a> PaimonTableRead<'a> {
) -> crate::Result<ArrowRecordBatchStream> {
use super::managed_blob_reader::{resolve_primary_key_blob_stream,
ManagedBlobReadPlan};
+ let data_splits = self.matching_splits(data_splits)?;
let core_options = self.table.schema().core_options();
let user_read_type = self.read_type.clone();
let audit_schema = audit_schema_for_read_type(&user_read_type,
include_sequence)?;
@@ -647,7 +695,7 @@ impl<'a> PaimonTableRead<'a> {
.with_parquet_read_budget(Some(self.parquet_read_budget()?))
.with_table_options(self.table.schema().options().clone())
.with_mosaic_prefetch(configured_mosaic_prefetch(self.table)?);
- let raw_stream = reader.read(data_splits)?;
+ let raw_stream = reader.read(&data_splits)?;
let raw_stream = if let Some(plan) = blob_plan {
plan.finish(
raw_stream,
@@ -724,6 +772,7 @@ impl<'a> PaimonTableRead<'a> {
let table = self.table.clone();
let read_type = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
+ let partition_predicate = self.partition_predicate.clone();
let parquet_read_budget = self.parquet_read_budget()?;
let blob_parallelism = self.blob_parallelism;
@@ -732,9 +781,11 @@ impl<'a> PaimonTableRead<'a> {
let table = table.clone();
let read_type = read_type.clone();
let data_predicates = data_predicates.clone();
+ let partition_predicate = partition_predicate.clone();
let parquet_read_budget = Arc::clone(&parquet_read_budget);
let worker: ArrowRecordBatchStream =
Box::pin(async_stream::try_stream! {
let pair_read = PaimonTableRead::new(&table, read_type,
data_predicates)
+ .with_partition_predicate(partition_predicate)
.with_parquet_read_budget(parquet_read_budget)
.with_blob_parallelism(blob_parallelism);
let mut pair_stream =
@@ -782,12 +833,14 @@ impl<'a> PaimonTableRead<'a> {
let table = self.table.clone();
let read_type_for_output = self.read_type.clone();
let data_predicates = self.data_predicates.clone();
+ let partition_predicate = self.partition_predicate.clone();
let parquet_read_budget = self.parquet_read_budget()?;
let blob_parallelism = self.blob_parallelism;
Ok(Box::pin(async_stream::try_stream! {
let core_options = CoreOptions::new(table.schema().options());
let pair_read = PaimonTableRead::new(&table,
diff_read_type.clone(), data_predicates)
+ .with_partition_predicate(partition_predicate)
.with_parquet_read_budget(parquet_read_budget)
.with_blob_parallelism(blob_parallelism);
let before_stream =
@@ -868,6 +921,7 @@ impl<'a> PaimonTableRead<'a> {
let table = self.table.clone();
let data_predicates = self.data_predicates.clone();
+ let partition_predicate = self.partition_predicate.clone();
let before = before.to_vec();
let after = after.to_vec();
let parquet_read_budget = self.parquet_read_budget()?;
@@ -876,6 +930,7 @@ impl<'a> PaimonTableRead<'a> {
Ok(Box::pin(async_stream::try_stream! {
let core_options = CoreOptions::new(table.schema().options());
let pair_read = PaimonTableRead::new(&table,
diff_read_type.clone(), data_predicates)
+ .with_partition_predicate(partition_predicate)
.with_parquet_read_budget(parquet_read_budget)
.with_blob_parallelism(blob_parallelism);
let before_stream = pair_read.read_pk_sorted_for_diff_with_type(
@@ -928,6 +983,7 @@ impl<'a> PaimonTableRead<'a> {
core_options: &CoreOptions,
read_type: &[DataField],
) -> crate::Result<ArrowRecordBatchStream> {
+ let splits = self.matching_splits(splits)?;
if splits.is_empty() {
return Ok(Box::pin(futures::stream::empty()));
}
@@ -962,7 +1018,7 @@ impl<'a> PaimonTableRead<'a> {
mosaic_prefetch: configured_mosaic_prefetch(self.table)?,
},
);
- reader.read(splits)
+ reader.read(&splits)
}
/// Every split must carry this handle's grant; returns it when it
restricts
@@ -1063,9 +1119,11 @@ impl<'a> PaimonTableRead<'a> {
/// Returns an [`ArrowRecordBatchStream`].
pub fn to_arrow(&self, data_splits: &[DataSplit]) ->
crate::Result<ArrowRecordBatchStream> {
let core_options = self.table.schema.core_options();
- match self.ensure_authorized_by_splits(&core_options, data_splits)? {
- Some(grant) => self.read_restricted(data_splits, &grant),
- None => self.read_splits(data_splits, &core_options),
+ let grant = self.ensure_authorized_by_splits(&core_options,
data_splits)?;
+ let data_splits = self.matching_splits(data_splits)?;
+ match grant {
+ Some(grant) => self.read_restricted(&data_splits, &grant),
+ None => self.read_splits(&data_splits, &core_options),
}
}
diff --git a/crates/paimon/tests/partition_filter_test.rs
b/crates/paimon/tests/partition_filter_test.rs
new file mode 100644
index 00000000..0ee258e0
--- /dev/null
+++ b/crates/paimon/tests/partition_filter_test.rs
@@ -0,0 +1,382 @@
+// 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.
+
+mod common;
+
+use arrow_array::{ArrayRef, Int32Array, RecordBatch, StringArray};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+use futures::TryStreamExt;
+use paimon::spec::{
+ DataType, Datum, IntType, Predicate, PredicateBuilder, Schema,
TableSchema, VarCharType,
+};
+use paimon::table::{IncrementalScanMode, Plan, ReadBuilder, Table};
+use std::sync::Arc;
+
+async fn partitioned(mode: &str) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("p", DataType::VarChar(VarCharType::string_type()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("region", DataType::VarChar(VarCharType::string_type()))
+ // Reverse partition order relative to the table's field order.
+ .partition_keys(["region", "p"]);
+ if mode == "pk" {
+ schema = schema
+ .primary_key(["id", "p", "region"])
+ .option("bucket", "1");
+ } else if mode == "de" {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ let path = format!("memory:/explicit-partitions-{mode}");
+ let (io, table) = memory_table(&path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, &path).await;
+ persist_table_schema(&io, &path, table.schema()).await;
+ write_batch(
+ &table,
+ &batch(&[
+ (1, "a", 10, "east"),
+ (2, "b", 20, "east"),
+ (3, "a", 30, "west"),
+ (4, "b", 40, "west"),
+ ]),
+ )
+ .await;
+ table
+}
+
+fn batch(rows: &[(i32, &str, i32, &str)]) -> RecordBatch {
+ RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from_iter_values(rows.iter().map(|r| r.0)))
as ArrayRef,
+ ),
+ (
+ "p",
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)))
as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from_iter_values(rows.iter().map(|r| r.2)))
as ArrayRef,
+ ),
+ (
+ "region",
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.3)))
as ArrayRef,
+ ),
+ ])
+ .unwrap()
+}
+
+fn equal(table: &Table, field: &str, value: &str) -> Predicate {
+ PredicateBuilder::new(table.schema().fields())
+ .equal(field, Datum::String(value.into()))
+ .unwrap()
+}
+
+async fn ids(builder: &ReadBuilder<'_>, plan: &Plan) -> Vec<i32> {
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut result = Vec::new();
+ for batch in batches {
+ assert_eq!(
+ batch.schema().fields().len(),
+ 1,
+ "selector/filter fields must stay hidden"
+ );
+ let column = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ result.extend(column.values());
+ }
+ result.sort_unstable();
+ result
+}
+
+#[tokio::test]
+async fn partition_selector_is_scan_only_and_independent_of_setter_order() {
+ for mode in ["append", "pk", "de"] {
+ let table = partitioned(mode).await;
+ for selector_first in [true, false] {
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ let row_filter = equal(&table, "p", "a");
+ let selector = equal(&table, "region", "east");
+ if selector_first {
+ builder
+ .with_partition_filter(Some(selector))
+ .unwrap()
+ .with_filter(row_filter);
+ } else {
+ builder
+ .with_filter(row_filter)
+ .with_partition_filter(Some(selector))
+ .unwrap();
+ }
+ let plan = builder.new_scan().plan().await.unwrap();
+ // Ordinary filters may further prune files by data statistics.
+ // The explicit selector must still restrict every planned
partition.
+ assert!(!plan.splits().is_empty(), "{mode}");
+ assert!(plan
+ .splits()
+ .iter()
+ .all(|s| s.partition().get_string(0).unwrap() == "east"));
+ assert_eq!(ids(&builder, &plan).await, [1], "{mode}");
+ // A separately configured reader must enforce its row predicate
+ // even though it was not given the scan-only selector.
+ let mut reader = table.new_read_builder();
+ reader
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(equal(&table, "p", "a"));
+ assert_eq!(ids(&reader, &plan).await, [1], "{mode}");
+ }
+ }
+}
+
+#[tokio::test]
+async fn replacing_and_clearing_a_selector_preserves_the_row_filter() {
+ for mode in ["append", "pk", "de"] {
+ let table = partitioned(mode).await;
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(equal(&table, "p", "a"));
+ builder
+ .with_partition_filter(Some(equal(&table, "region", "east")))
+ .unwrap();
+ let captured = builder.clone();
+ builder
+ .with_partition_filter(Some(equal(&table, "region", "west")))
+ .unwrap();
+ assert_eq!(
+ ids(&builder, &builder.new_scan().plan().await.unwrap()).await,
+ [3]
+ );
+ assert_eq!(
+ ids(&captured, &captured.new_scan().plan().await.unwrap()).await,
+ [1]
+ );
+ builder.with_partition_filter(None).unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert_eq!(plan.splits().len(), 2);
+ assert_eq!(ids(&builder, &plan).await, [1, 3]);
+ // A later ordinary-filter setter does not reset the explicit selector.
+ builder
+ .with_partition_filter(Some(equal(&table, "region", "west")))
+ .unwrap()
+ .with_filter(equal(&table, "p", "b"));
+ assert_eq!(
+ ids(&builder, &builder.new_scan().plan().await.unwrap()).await,
+ [4]
+ );
+ }
+}
+
+#[tokio::test]
+async fn
contradictory_partition_predicates_do_not_return_rows_or_truncate_matches() {
+ for mode in ["append", "pk", "de"] {
+ let table = partitioned(mode).await;
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id"])
+ .unwrap()
+ .with_limit(1)
+ .with_filter(equal(&table, "region", "west"));
+ builder
+ .with_partition_filter(Some(equal(&table, "region", "east")))
+ .unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ assert!(ids(&builder, &plan).await.is_empty());
+ // A broader selector plus LIMIT cannot count the east files before a
+ // west row has survived the ordinary partition predicate.
+ builder
+ .with_partition_filter(Some(Predicate::or(vec![
+ equal(&table, "region", "east"),
+ equal(&table, "region", "west"),
+ ])))
+ .unwrap();
+ let result = ids(&builder,
&builder.new_scan().plan().await.unwrap()).await;
+ assert!(!result.is_empty());
+ assert!(result.iter().all(|id| [3, 4].contains(id)));
+ // Append/PK callers enforce the final row count; DE enforces it in
core.
+ if mode == "de" {
+ assert_eq!(result.len(), 1);
+ }
+ }
+}
+
+#[tokio::test]
+async fn selector_survives_incremental_stream_checkpoint_and_restore() {
+ for mode in ["append", "pk", "de"] {
+ let table = partitioned(mode).await;
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ builder
+ .with_partition_filter(Some(equal(&table, "region", "east")))
+ .unwrap();
+ let mut stream = builder.new_stream_scan().unwrap();
+ assert_eq!(
+ ids(&builder, &stream.plan().await.unwrap().unwrap()).await,
+ [1, 2]
+ );
+ assert_eq!(stream.checkpoint(), Some(2));
+ write_batch(
+ &table,
+ &batch(&[(5, "a", 50, "west"), (6, "b", 60, "east")]),
+ )
+ .await;
+ let delta = builder
+ .new_incremental_scan(IncrementalScanMode::Delta, 1, 2)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(ids(&builder, &delta).await, [6]);
+ assert_eq!(
+ ids(&builder, &stream.plan().await.unwrap().unwrap()).await,
+ [6]
+ );
+ stream.restore(Some(2)).unwrap();
+ assert_eq!(
+ ids(&builder, &stream.plan().await.unwrap().unwrap()).await,
+ [6]
+ );
+ }
+}
+
+#[tokio::test]
+async fn invalid_selector_fails_before_mutating_the_previous_selection() {
+ let table = partitioned("append").await;
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ builder
+ .with_partition_filter(Some(equal(&table, "region", "east")))
+ .unwrap();
+ let nonpartition = PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(3))
+ .unwrap();
+ for invalid in [
+ nonpartition.clone(),
+ Predicate::or(vec![equal(&table, "p", "a"), nonpartition]),
+ ] {
+ let error = builder.with_partition_filter(Some(invalid)).unwrap_err();
+ assert!(error.to_string().contains("only partition keys"));
+ assert_eq!(
+ ids(&builder, &builder.new_scan().plan().await.unwrap()).await,
+ [1, 2]
+ );
+ }
+}
+
+#[tokio::test]
+async fn direct_table_read_filter_selects_independently_planned_partitions() {
+ for mode in ["append", "pk", "de"] {
+ let table = partitioned(mode).await;
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap();
+ let read = builder
+ .new_read()
+ .unwrap()
+ .with_filter(equal(&table, "p", "a"));
+ let batches = read
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut actual = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ actual.sort_unstable();
+ assert_eq!(actual, [1, 3], "{mode}");
+ }
+}
+
+#[tokio::test]
+async fn ordinary_partition_filter_survives_diff_workers_and_audit_images() {
+ let table = partitioned("pk").await;
+ write_batch(
+ &table,
+ &batch(&[(1, "a", 100, "east"), (2, "b", 200, "east")]),
+ )
+ .await;
+ let plan = table
+ .new_read_builder()
+ .new_incremental_scan(IncrementalScanMode::Diff, 1, 2)
+ .plan()
+ .await
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(equal(&table, "p", "a"));
+ let read = builder.new_read().unwrap();
+ let rows = read
+ .to_incremental_arrow(&plan)
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(rows.iter().map(|batch| batch.num_rows()).sum::<usize>(), 1);
+ assert_eq!(
+ rows[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 1
+ );
+ let audit = read
+ .to_audit_log_arrow(&plan)
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(audit.iter().map(|batch| batch.num_rows()).sum::<usize>(), 2);
+ for batch in audit {
+ assert!(batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .all(|id| *id == 1));
+ }
+}
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 5a71d478..8251943b 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -5719,3 +5719,114 @@ async fn
test_branch_write_copy_rejects_replaced_source_identity_or_location() {
"{error}"
);
}
+
+/// Explicit selectors are scan-only and must not turn into masked row filters.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_query_auth_explicit_partition_selector() {
+ for data_evolution in [false, true] {
+ let schema = |options: &[(&str, &str)]| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("dept",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .column("region",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .partition_keys(["region", "dept"]);
+ if data_evolution {
+ builder = builder
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ };
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as
arrow_array::ArrayRef,
+ ),
+ (
+ "dept",
+ Arc::new(StringArray::from(vec!["eng", "ops", "eng", "ops"]))
+ as arrow_array::ArrayRef,
+ ),
+ (
+ "region",
+ Arc::new(StringArray::from(vec!["east", "east", "west",
"west"]))
+ as arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let g = written("explicit_parts", schema, vec![batch]).await;
+ let predicates = PredicateBuilder::new(g.table.schema().fields());
+ let equal = |field: &str, value: &str| {
+ predicates
+ .equal(field, Datum::String(value.into()))
+ .unwrap()
+ };
+ g.set_rules(&[], &[("dept", upper(1, "dept"))]);
+ let mut read = g.table.new_read_builder();
+ read.with_projection(&["id"])
+ .unwrap()
+ .with_filter(equal("dept", "ENG"))
+ .with_partition_filter(Some(equal("region", "east")))
+ .unwrap()
+ .with_limit(1);
+ let plan = read.new_scan().plan().await.unwrap();
+ assert_eq!(plan.splits().len(), 2);
+ assert_eq!(ids(&read_all(&read).await.unwrap()), [1]);
+ let selected = g.ctx.server.auth_selects("default", "explicit_parts");
+ assert!(selected.iter().all(|select| {
+ let mut fields = select.clone().unwrap();
+ fields.sort();
+ fields == ["dept", "id", "region"]
+ }));
+
+ // A mask on another partition key must not reject an unmasked
selector.
+ let mut selector_only = g.table.new_read_builder();
+ selector_only
+ .with_projection(&["id"])
+ .unwrap()
+ .with_partition_filter(Some(equal("region", "west")))
+ .unwrap();
+ assert_eq!(ids(&read_all(&selector_only).await.unwrap()), [3, 4]);
+
+ // A selector on a masked key is unsafe even with an empty
result/LIMIT 0.
+ for limit in [0, 1] {
+ let mut masked = g.table.new_read_builder();
+ masked
+ .with_projection(&["id"])
+ .unwrap()
+ .with_limit(limit)
+ .with_partition_filter(Some(equal("dept", "missing")))
+ .unwrap();
+ let error = masked.new_scan().plan().await.unwrap_err();
+ assert!(
+ error.to_string().contains("partition key 'dept' is masked"),
+ "{error}"
+ );
+ }
+ // Validate names before equality predicates become opaque partition
sets.
+ let forged = Predicate::Leaf {
+ column: "id".into(),
+ index: 1,
+ data_type: g.table.schema().fields()[1].data_type().clone(),
+ op: PredicateOperator::Eq,
+ literals: vec![Datum::String("eng".into())],
+ };
+ assert_disagreeing_leaf(
+ g.table
+ .new_read_builder()
+ .with_partition_filter(Some(forged))
+ .unwrap_err(),
+ );
+
+ // An explicit scan-only operand must be included in column
authorization.
+ g.set_rules(&[], &[]);
+ g.ctx
+ .server
+ .set_column_auth("default", "explicit_parts", vec!["id".into()]);
+ assert_forbidden(selector_only.new_scan().plan().await.unwrap_err());
+ }
+}
