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 c33f79d5 [core] Support authorized raw file planning for native scans
(#1197)
c33f79d5 is described below
commit c33f79d5b1c2e8234d4c1408405e53a9a0807e5f
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 22 16:34:29 2026 +0800
[core] Support authorized raw file planning for native scans (#1197)
---
.../python/python/pypaimon_rust/datafusion.pyi | 19 +-
bindings/python/src/commit_callback.rs | 6 +
bindings/python/src/context.rs | 1 +
bindings/python/src/read.rs | 260 ++++++++++++++++++++-
crates/paimon/src/table/authorized_update_tests.rs | 12 +-
crates/paimon/src/table/data_evolution_writer.rs | 6 +-
crates/paimon/src/table/incremental_scan.rs | 26 ++-
crates/paimon/src/table/mod.rs | 2 +-
crates/paimon/src/table/shard_table_updater.rs | 2 +-
crates/paimon/src/table/snapshot_scan_tests.rs | 235 +++++++++++++++++++
crates/paimon/src/table/table_merge_input.rs | 2 +-
crates/paimon/src/table/table_merge_into.rs | 2 +-
crates/paimon/src/table/table_scan.rs | 243 ++++++++++++++++++-
crates/paimon/src/table/table_scan/file_scan.rs | 234 +++++++++++++++++++
crates/paimon/src/table/table_upsert.rs | 2 +-
.../tests/incremental_timestamp_scan_test.rs | 41 ++++
crates/paimon/tests/nullable_primary_key_test.rs | 2 +-
crates/paimon/tests/rest_catalog/merge_sources.rs | 2 +-
.../tests/rest_catalog/read_path_completion.rs | 118 +++++++++-
crates/paimon/tests/rest_catalog_test.rs | 2 +-
20 files changed, 1175 insertions(+), 42 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 81770d15..ad573e03 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -164,6 +164,11 @@ class GlobalIndexResult:
...
class TableScan:
+ def with_file_filters(self, skip_level0: bool, only_read_real_buckets:
bool) -> TableScan:
+ """Set independent level-zero and pending-bucket policies for
+ plan() and plan_files(). plan_all() includes L0 and retains the
+ configured bucket policy."""
+ ...
def with_global_index_result(self, result: GlobalIndexResult) ->
"TableScan":
"""Reuse an index selection. Authorized scans ignore supplied index
selections."""
...
@@ -179,10 +184,20 @@ class TableScan:
ranges of shuffled chunks. Row-position sharding is a separate API."""
...
def plan(self) -> Plan: ...
- def plan_all_files(self) -> Plan:
- """Include L0 and all column files, retaining scan filters and limits.
Requires unrestricted query authorization."""
+ def plan_all(self) -> Plan:
+ """Split Plan including L0 and all matching data-evolution column
files.
+ Retains scan filters and limits. Requires unrestricted query
authorization."""
+ ...
+ def plan_files(self) -> FileScanPlan:
+ """Filtered file entries before projection, LIMIT, row selection and
index refinement.
+ Without explicit file filters, includes level zero and pending buckets.
+ Requires unrestricted whole-table query authorization."""
...
+class FileScanPlan:
+ def snapshot_id(self) -> Optional[int]: ...
+ def files(self) -> List[ManifestEntry]: ...
+
class StreamTableScan:
"""Stateful continuous scan; checkpoint is the next snapshot to consume."""
diff --git a/bindings/python/src/commit_callback.rs
b/bindings/python/src/commit_callback.rs
index 7bb744c3..0dbe0a71 100644
--- a/bindings/python/src/commit_callback.rs
+++ b/bindings/python/src/commit_callback.rs
@@ -80,6 +80,12 @@ pub struct PyManifestEntry {
inner: ManifestEntry,
}
+impl From<ManifestEntry> for PyManifestEntry {
+ fn from(inner: ManifestEntry) -> Self {
+ Self { inner }
+ }
+}
+
#[pymethods]
impl PyManifestEntry {
#[getter]
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 102a3a4a..7e56ceda 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -1019,6 +1019,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<crate::read::PyTableScan>()?;
this.add_class::<crate::read::PyStreamTableScan>()?;
this.add_class::<crate::read::PyPlan>()?;
+ this.add_class::<crate::read::PyFileScanPlan>()?;
this.add_class::<crate::read::PyScanStats>()?;
this.add_class::<crate::read::PyTableRead>()?;
this.add_class::<crate::read::PyRecordBatchReader>()?;
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 30ffdf52..05996167 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -351,6 +351,7 @@ impl PyReadBuilder {
row_position_shard: None,
chunk_shuffle: None,
shard: None,
+ file_filters: None,
}
}
@@ -415,6 +416,7 @@ pub struct PyTableScan {
row_position_shard: Option<(u64, u64)>,
chunk_shuffle: Option<PyChunkShuffle>,
shard: Option<(usize, usize)>,
+ file_filters: Option<(bool, bool)>,
}
#[derive(Clone)]
@@ -470,6 +472,11 @@ impl PyTableScan {
if let Some((index, count)) = self.shard {
scan = scan.with_shard(index, count).map_err(to_py_err)?;
}
+ if let Some((skip_level0, only_read_real_buckets)) = self.file_filters
{
+ scan = scan
+ .with_file_filters(skip_level0, only_read_real_buckets)
+ .map_err(to_py_err)?;
+ }
Ok(scan)
}
@@ -691,6 +698,25 @@ impl PyStreamTableScan {
#[pymethods]
impl PyTableScan {
+ /// Configure independent file filters for plan() and plan_files().
+ /// plan_all() includes L0 while retaining the configured bucket policy.
+ fn with_file_filters(
+ mut slf: PyRefMut<'_, Self>,
+ skip_level0: bool,
+ only_read_real_buckets: bool,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ if slf.incremental_scan.is_some() {
+ return Err(PyValueError::new_err(
+ "file filters use table snapshot or
incremental-between-timestamp options",
+ ));
+ }
+ slf.core_scan()?
+ .with_file_filters(skip_level0, only_read_real_buckets)
+ .map_err(to_py_err)?;
+ slf.file_filters = Some((skip_level0, only_read_real_buckets));
+ Ok(slf)
+ }
+
/// Supply an evaluated selection; validation and authorization stay in
core.
fn with_global_index_result<'py>(
mut slf: PyRefMut<'py, Self>,
@@ -769,11 +795,13 @@ impl PyTableScan {
Ok(slf)
}
- fn plan_all_files(&self, py: Python<'_>) -> PyResult<PyPlan> {
+ /// Return a split Plan including L0 and all matching data-evolution
column files.
+ /// Scan filters and limits remain in effect; requires unrestricted query
authorization.
+ fn plan_all(&self, py: Python<'_>) -> PyResult<PyPlan> {
py.detach(|| {
runtime().block_on(async {
self.core_scan()?
- .plan_all_files()
+ .plan_all()
.await
.map(PyPlan::from)
.map_err(to_py_err)
@@ -781,6 +809,29 @@ impl PyTableScan {
})
}
+ /// Return filtered file metadata before split packing or row selection.
+ fn plan_files(&self, py: Python<'_>) -> PyResult<PyFileScanPlan> {
+ if self.incremental_scan.is_some() {
+ return Err(PyValueError::new_err(
+ "plan_files uses table snapshot or
incremental-between-timestamp options",
+ ));
+ }
+ py.detach(|| {
+ runtime().block_on(async {
+ let mut scan = self.read_builder()?.new_scan();
+ if let Some((skip_level0, only_read_real_buckets)) =
self.file_filters {
+ scan = scan
+ .with_file_filters(skip_level0, only_read_real_buckets)
+ .map_err(to_py_err)?;
+ }
+ scan.plan_files()
+ .await
+ .map(|inner| PyFileScanPlan { inner })
+ .map_err(to_py_err)
+ })
+ })
+ }
+
fn plan(&self, py: Python<'_>) -> PyResult<PyPlan> {
py.detach(|| {
runtime().block_on(async {
@@ -1122,6 +1173,22 @@ pub struct PyPlan {
inner: paimon::table::Plan,
}
+#[pyclass(name = "FileScanPlan", module = "pypaimon_rust.datafusion", frozen)]
+pub struct PyFileScanPlan {
+ inner: paimon::table::FileScanPlan,
+}
+
+#[pymethods]
+impl PyFileScanPlan {
+ fn snapshot_id(&self) -> Option<i64> {
+ self.inner.snapshot_id()
+ }
+
+ fn files(&self) -> Vec<crate::commit_callback::PyManifestEntry> {
+ self.inner.files().iter().cloned().map(Into::into).collect()
+ }
+}
+
impl From<paimon::table::Plan> for PyPlan {
fn from(plan: paimon::table::Plan) -> Self {
Self { inner: plan }
@@ -1242,6 +1309,195 @@ mod tests {
use super::*;
+ #[test]
+ fn
file_planning_binding_keeps_complete_entries_and_independent_query_filters() {
+ Python::attach(|py| {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", "first-row")
+ .build()
+ .unwrap();
+ let table = Arc::new(Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "file_binding"),
+ "memory:/file-binding".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ py.detach(|| {
+ runtime().block_on(async {
+ table
+ .file_io()
+ .mkdirs("memory:/file-binding/schema")
+ .await
+ .unwrap();
+ table
+ .file_io()
+ .mkdirs("memory:/file-binding/snapshot")
+ .await
+ .unwrap();
+ table
+ .file_io()
+ .new_output("memory:/file-binding/schema/schema-0")
+ .unwrap()
+
.write(serde_json::to_vec(table.schema()).unwrap().into())
+ .await
+ .unwrap();
+ for id in [1, 1] {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch = RecordBatch::try_from_iter(vec![(
+ "id",
+ Arc::new(arrow::array::Int32Array::from(vec![id]))
+ as arrow::array::ArrayRef,
+ )])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ writer.close().await;
+ }
+ })
+ });
+ let mut builder = PyReadBuilder::new(Arc::clone(&table));
+ builder.limit = Some(0);
+ let scan = Py::new(py, builder.new_scan()).unwrap();
+ for name in ["skip_level0", "only_read_real_buckets"] {
+ let kwargs = PyDict::new(py);
+ kwargs.set_item(name, true).unwrap();
+ let error = scan
+ .bind(py)
+ .call_method("plan_files", (), Some(&kwargs))
+ .unwrap_err();
+
assert!(error.is_instance_of::<pyo3::exceptions::PyTypeError>(py));
+ }
+ assert_eq!(
+ scan.bind(py)
+ .call_method0("plan_all")
+ .unwrap()
+ .len()
+ .unwrap(),
+ 0,
+ "plan_all retains row LIMIT"
+ );
+ let files = scan.bind(py).call_method0("plan_files").unwrap();
+ assert_eq!(
+ files
+ .call_method0("snapshot_id")
+ .unwrap()
+ .extract::<i64>()
+ .unwrap(),
+ 2
+ );
+ let entries = files.call_method0("files").unwrap();
+ assert_eq!(
+ entries.len().unwrap(),
+ 2,
+ "metadata planning ignores row LIMIT"
+ );
+ let first = entries.get_item(0).unwrap();
+
assert_eq!(first.getattr("kind").unwrap().extract::<i32>().unwrap(), 0);
+ assert_eq!(
+ first
+ .getattr("total_buckets")
+ .unwrap()
+ .extract::<i32>()
+ .unwrap(),
+ 1
+ );
+ let file = first.getattr("file").unwrap();
+ assert_eq!(
+ file.getattr("row_count").unwrap().extract::<i64>().unwrap(),
+ 1
+ );
+ assert!(file
+ .getattr("file_name")
+ .unwrap()
+ .extract::<String>()
+ .unwrap()
+ .ends_with(".parquet"));
+ let encoded = first
+ .call_method0("_to_commit_message")
+ .unwrap()
+ .call_method0("serialize")
+ .unwrap();
+ let encoded = encoded.cast::<PyBytes>().unwrap();
+ let restored = paimon::table::CommitMessage::deserialize(
+ paimon::table::COMMIT_MESSAGE_SERIALIZER_VERSION,
+ encoded.as_bytes(),
+ )
+ .unwrap();
+ let expected = py.detach(|| {
+ runtime().block_on(async {
+ table
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .unwrap()
+ })
+ });
+ assert_eq!(restored.new_files,
vec![expected.files()[0].file().clone()]);
+
+ builder.limit = None;
+ let scan = Py::new(py, builder.new_scan()).unwrap();
+ assert_eq!(
+ scan.bind(py).call_method0("plan").unwrap().len().unwrap(),
+ 0
+ );
+ scan.bind(py)
+ .call_method1("with_file_filters", (false, true))
+ .unwrap();
+ assert_eq!(
+ scan.bind(py).call_method0("plan").unwrap().len().unwrap(),
+ 1
+ );
+ for (skip_level0, expected_files, expected_splits) in
+ [(false, 2, 1), (true, 0, 0), (false, 2, 1)]
+ {
+ scan.bind(py)
+ .call_method1("with_file_filters", (skip_level0, true))
+ .unwrap();
+ assert_eq!(
+ scan.bind(py)
+ .call_method0("plan_files")
+ .unwrap()
+ .call_method0("files")
+ .unwrap()
+ .len()
+ .unwrap(),
+ expected_files
+ );
+ assert_eq!(
+ scan.bind(py).call_method0("plan").unwrap().len().unwrap(),
+ expected_splits
+ );
+ let all = scan.bind(py).call_method0("plan_all").unwrap();
+ let all = all.extract::<PyRef<'_, PyPlan>>().unwrap();
+ assert_eq!(
+ all.inner
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .count(),
+ 2,
+ "plan_all includes L0 even when explicit file filters
exclude it"
+ );
+ }
+ let fresh = Py::new(py, builder.new_scan()).unwrap();
+ assert_eq!(
+ fresh.bind(py).call_method0("plan").unwrap().len().unwrap(),
+ 0,
+ "file filters must not change a fresh query scan's defaults"
+ );
+ });
+ }
+
#[test]
fn stream_scan_copy_keeps_callback_ownership_and_progress_independent() {
Python::attach(|py| {
diff --git a/crates/paimon/src/table/authorized_update_tests.rs
b/crates/paimon/src/table/authorized_update_tests.rs
index aa9d8cb6..eb24ea09 100644
--- a/crates/paimon/src/table/authorized_update_tests.rs
+++ b/crates/paimon/src/table/authorized_update_tests.rs
@@ -353,7 +353,7 @@ async fn
assembled_handles_cannot_mutate_or_replay_a_live_grant() {
.table
.new_read_builder()
.new_scan()
- .plan_all_files()
+ .plan_all()
.await
.unwrap();
let mut builder = fixture.table.new_read_builder();
@@ -397,7 +397,7 @@ async fn
only_unrestricted_live_grants_allow_internal_row_id_reads() {
.table
.new_read_builder()
.new_scan()
- .plan_all_files()
+ .plan_all()
.await
.unwrap();
let serialized = serde_json::to_vec(plan.splits()).unwrap();
@@ -503,7 +503,7 @@ async fn
mutation_metadata_predicates_match_unguarded_and_authorized_rows() {
.table
.new_read_builder()
.new_scan()
- .plan_all_files()
+ .plan_all()
.await
.unwrap();
let first = plan
@@ -624,14 +624,14 @@ async fn
mutation_metadata_predicates_preserve_restricted_and_public_read_guards
.unwrap(),
);
assert!(builder.new_scan().plan().await.is_err());
- assert!(builder.new_scan().plan_all_files().await.is_err());
+ assert!(builder.new_scan().plan_all().await.is_err());
}
assert!(fixture
.table
.new_read_builder()
.new_scan()
.with_row_ranges(vec![super::super::RowRange::new(0, 0)])
- .plan_all_files()
+ .plan_all()
.await
.is_err());
assert_eq!(fixture.state.query_requests.load(Ordering::SeqCst), 0);
@@ -639,7 +639,7 @@ async fn
mutation_metadata_predicates_preserve_restricted_and_public_read_guards
.table
.new_read_builder()
.new_scan()
- .plan_all_files()
+ .plan_all()
.await
.unwrap();
// The internal marker allows canonical _ROW_ID output only. A forged
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 38c5aaa4..2df6a785 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -612,11 +612,7 @@ impl DataEvolutionDeleteWriter {
// A row filter or mask cannot authorize a physical file mutation.
let snapshot =
super::time_travel::resolve_snapshot(&self.table).await?;
let scan_table =
self.table.copy_with_pinned_snapshot(snapshot.as_ref());
- let plan = scan_table
- .new_read_builder()
- .new_scan()
- .plan_all_files()
- .await?;
+ let plan = scan_table.new_read_builder().new_scan().plan_all().await?;
let snapshot = snapshot.ok_or_else(|| crate::Error::DataInvalid {
message: "No files with row tracking found in target table".into(),
source: None,
diff --git a/crates/paimon/src/table/incremental_scan.rs
b/crates/paimon/src/table/incremental_scan.rs
index 70ece922..d06b4c03 100644
--- a/crates/paimon/src/table/incremental_scan.rs
+++ b/crates/paimon/src/table/incremental_scan.rs
@@ -423,6 +423,27 @@ impl<'a> IncrementalScan<'a> {
// valid boundary even though that snapshot need not exist.
return
Ok(Plan::new(Vec::new()).with_snapshot_id(self.end_inclusive));
}
+ let (end_snapshot, manifest_lists) =
self.selected_event_manifest_lists().await?;
+ let plan = scan
+ .plan_incremental_lists(&end_snapshot, manifest_lists, scan_stats)
+ .await?;
+ self.scan.attach_global_index_scores(plan)
+ }
+
+ /// Collect the physical event sources after the caller resolves and
validates
+ /// the snapshot range. File metadata scans use the same AUTO selection as
rows.
+ pub(crate) async fn selected_event_manifest_lists(
+ &self,
+ ) -> crate::Result<(crate::spec::Snapshot, Vec<String>)> {
+ let mode = self.resolve_mode();
+ if !matches!(
+ mode,
+ IncrementalScanMode::Delta | IncrementalScanMode::Changelog
+ ) {
+ return Err(crate::Error::Unsupported {
+ message: "File event sources require Delta or Changelog
mode".into(),
+ });
+ }
let end_snapshot = self
.snapshot_manager
.get_snapshot(self.end_inclusive)
@@ -454,10 +475,7 @@ impl<'a> IncrementalScan<'a> {
})
.map(str::to_owned)
.collect();
- let plan = scan
- .plan_incremental_lists(&end_snapshot, manifest_lists, scan_stats)
- .await?;
- self.scan.attach_global_index_scores(plan)
+ Ok((end_snapshot, manifest_lists))
}
fn resolve_mode(&self) -> IncrementalScanMode {
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 7cbce5ba..0fee5af8 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -260,7 +260,7 @@ pub use table_merge_into::{
MergeAssignment, MergeCondition, MergeSource, WhenMatched, WhenNotMatched,
};
pub use table_read::{AuditLogRead, TableRead};
-pub use table_scan::{AuditLogScan, TableScan};
+pub use table_scan::{AuditLogScan, FileScanPlan, TableScan};
pub use table_update::TableUpdate;
pub use table_update_by_row_id::TableUpdateByRowId;
pub use table_write::TableWrite;
diff --git a/crates/paimon/src/table/shard_table_updater.rs
b/crates/paimon/src/table/shard_table_updater.rs
index 3872af02..36695190 100644
--- a/crates/paimon/src/table/shard_table_updater.rs
+++ b/crates/paimon/src/table/shard_table_updater.rs
@@ -99,7 +99,7 @@ impl ShardTableUpdater {
"FULL".into(),
)]))
.copy_with_pinned_snapshot(snapshot.as_ref());
- let plan = table.new_read_builder().new_scan().plan_all_files().await?;
+ let plan = table.new_read_builder().new_scan().plan_all().await?;
let splits = select_shard(plan.splits(), shard, shard_count)?;
let index = RowIdFileIndex::from_splits(table.clone(), &splits)?;
let read_type = read_type.unwrap_or_else(||
table.schema().fields().to_vec());
diff --git a/crates/paimon/src/table/snapshot_scan_tests.rs
b/crates/paimon/src/table/snapshot_scan_tests.rs
index c535dc2a..74fe00a4 100644
--- a/crates/paimon/src/table/snapshot_scan_tests.rs
+++ b/crates/paimon/src/table/snapshot_scan_tests.rs
@@ -665,3 +665,238 @@ async fn
postpone_pending_visibility_is_an_explicit_reader_selection() {
}
}
}
+
+#[tokio::test]
+async fn explicit_file_filters_include_level_zero_with_key_merge() {
+ for engine in ["first-row", "deduplicate"] {
+ for buckets in [1, -2] {
+ let original = table(
+ &format!("file_filters_{engine}_{buckets}"),
+ engine,
+ "none",
+ buckets,
+ )
+ .await;
+ // DV's default read hides unmerged level zero. Explicit inclusion
must
+ // pack overlapping keys together and let the reader merge their
versions.
+ let table = if engine == "deduplicate" {
+ original.copy_with_options(HashMap::from([(
+ "deletion-vectors.enabled".into(),
+ "true".into(),
+ )]))
+ } else {
+ original
+ };
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ write_batch(&table, &make_batch(vec![1], vec![20])).await;
+ let builder = table.new_read_builder();
+ let default = builder.new_scan().plan().await.unwrap();
+ assert!(
+ default.splits().is_empty(),
+ "{engine} must keep its ordinary L0 exclusion"
+ );
+ let scan = builder
+ .new_scan()
+ .with_file_filters(false, buckets > 0)
+ .unwrap();
+ assert_eq!(scan.plan_files().await.unwrap().files().len(), 2);
+ let selected = scan.plan().await.unwrap();
+ assert_eq!(
+ selected.splits().len(),
+ 1,
+ "overlapping versions belong together"
+ );
+ assert!(!selected.splits()[0].raw_convertible());
+ let expected = if engine == "first-row" { 10 } else { 20 };
+ assert_eq!(rows(&table, &selected).await, [(1, expected)]);
+ for index in 0..3 {
+ let shard = builder
+ .new_scan()
+ .with_file_filters(false, buckets > 0)
+ .unwrap()
+ .with_shard(index, 3)
+ .unwrap()
+ .plan()
+ .await
+ .unwrap();
+ let owner = if buckets < 0 { 2 } else { 0 };
+ let expected_rows = if index == owner {
+ vec![(1, expected)]
+ } else {
+ vec![]
+ };
+ assert_eq!(
+ rows(&table, &shard).await,
+ expected_rows,
+ "{engine} bucket={buckets} worker={index}"
+ );
+ }
+ let all = builder
+ .new_scan()
+ .with_file_filters(true, true)
+ .unwrap()
+ .plan_all()
+ .await
+ .unwrap();
+ let baseline = builder.new_scan().plan_all().await.unwrap();
+ if buckets > 0 {
+ assert_eq!(all.splits(), baseline.splits());
+ } else {
+ assert!(
+ all.splits().is_empty(),
+ "whole-file planning retains the independent real-buckets
filter"
+ );
+ }
+ assert_eq!(
+ baseline
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .count(),
+ 2
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn file_scan_preserves_compacted_key_merge_inputs() {
+ for engine in ["deduplicate", "partial-update", "aggregation"] {
+ let table = table(&format!("file_merge_inputs_{engine}"), engine,
"none", 1).await;
+ for (level, value) in [(2, 10), (1, 20)] {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![1], vec![value]))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ for message in &mut messages {
+ for file in &mut message.new_files {
+ file.level = level;
+ }
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ let expected_value = if engine == "aggregation" { 30 } else { 20 };
+ let mut builder = table.new_read_builder();
+ builder.with_filter(
+ paimon::spec::PredicateBuilder::new(table.schema().fields())
+ .equal("value", paimon::spec::Datum::Int(expected_value))
+ .unwrap(),
+ );
+ for filters in [None, Some((false, false)), Some((true, false))] {
+ let mut scan = builder.new_scan();
+ if let Some((skip, only_real)) = filters {
+ scan = scan.with_file_filters(skip, only_real).unwrap();
+ }
+ let plan = scan.plan().await.unwrap();
+ assert_eq!(rows(&table, &plan).await, [(1, expected_value)]);
+ let files = scan.plan_files().await.unwrap();
+ assert_eq!(
+ files.files().len(),
+ 2,
+ "{engine} {filters:?}: skipping L0 must retain overlapping
L1/L2 inputs"
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn file_scan_uses_effective_first_row_level_policy_for_value_stats() {
+ let uncompacted = table("file_first_row_value_stats", "first-row", "none",
1).await;
+ write_batch(&uncompacted, &make_batch(vec![1], vec![10])).await;
+ write_batch(&uncompacted, &make_batch(vec![1], vec![20])).await;
+ let mut builder = uncompacted.new_read_builder();
+ builder.with_filter(
+ paimon::spec::PredicateBuilder::new(uncompacted.schema().fields())
+ .equal("value", paimon::spec::Datum::Int(20))
+ .unwrap(),
+ );
+ assert_eq!(
+ builder.new_scan().plan_files().await.unwrap().files().len(),
+ 2,
+ "default raw planning includes L0 merge inputs, unlike a normal
first-row query"
+ );
+ assert!(builder
+ .new_scan()
+ .with_file_filters(true, false)
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .files()
+ .is_empty());
+
+ let compacted = table("file_first_row_materialized", "first-row", "none",
1).await;
+ let builder = compacted.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![2], vec![20]))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ messages[0].new_files[0].level = 1;
+ builder.new_commit().commit(messages).await.unwrap();
+ let mut builder = compacted.new_read_builder();
+ builder.with_filter(
+ paimon::spec::PredicateBuilder::new(compacted.schema().fields())
+ .equal("value", paimon::spec::Datum::Int(10))
+ .unwrap(),
+ );
+ assert!(
+ builder
+ .new_scan()
+ .with_file_filters(true, false)
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap()
+ .files()
+ .is_empty(),
+ "materialized first-row files still allow value stats pruning"
+ );
+}
+
+#[tokio::test]
+async fn plan_all_preserves_independent_pending_bucket_filters() {
+ let table = table("all_pending_file_filters", "deduplicate", "none",
-2).await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let builder = table.new_read_builder();
+ let baseline = builder.new_scan().plan_all().await.unwrap();
+ assert_eq!(baseline.splits().len(), 1);
+ assert!(baseline.splits()[0].bucket() < 0);
+ for skip_level0 in [false, true] {
+ let scan = builder
+ .new_scan()
+ .with_file_filters(skip_level0, true)
+ .unwrap();
+ assert!(scan.plan().await.unwrap().splits().is_empty());
+ assert!(scan.plan_files().await.unwrap().files().is_empty());
+ assert!(scan
+ .clone()
+ .with_scan_all_files()
+ .plan_files()
+ .await
+ .unwrap()
+ .files()
+ .is_empty());
+ assert!(
+ scan.plan_all().await.unwrap().splits().is_empty(),
+ "plan_all disables L0 filtering without exposing excluded pending
buckets"
+ );
+ }
+ let all = builder
+ .new_scan()
+ .only_read_snapshot_real_buckets()
+ .with_file_filters(true, false)
+ .unwrap()
+ .plan_all()
+ .await
+ .unwrap();
+ assert_eq!(
+ all.splits(),
+ baseline.splits(),
+ "an explicit all-buckets policy replaces a previous real-buckets
selection"
+ );
+}
diff --git a/crates/paimon/src/table/table_merge_input.rs
b/crates/paimon/src/table/table_merge_input.rs
index 8c95ee07..ffdc71c6 100644
--- a/crates/paimon/src/table/table_merge_input.rs
+++ b/crates/paimon/src/table/table_merge_input.rs
@@ -121,7 +121,7 @@ impl MergeInput {
};
// Pin the source independently, preserving its resolved read schema.
// Authorized sources keep their original catalog identity and plan the
- // selected snapshot explicitly below. Include L0 as plan_all_files.
+ // selected snapshot explicitly below. Include L0 as plan_all.
let query_auth = table.schema().core_options().query_auth_enabled();
let table = if incremental || query_auth {
table.clone()
diff --git a/crates/paimon/src/table/table_merge_into.rs
b/crates/paimon/src/table/table_merge_into.rs
index 4aa129db..4abdff74 100644
--- a/crates/paimon/src/table/table_merge_into.rs
+++ b/crates/paimon/src/table/table_merge_into.rs
@@ -304,7 +304,7 @@ pub(super) async fn merge_into(
builder.with_projection(&projection.iter().map(String::as_str).collect::<Vec<_>>())?;
// The target may be empty, but its whole-table authorization must still
// precede source callbacks and insert-only MERGE actions.
- let plan = builder.new_scan().plan_all_files().await?;
+ let plan = builder.new_scan().plan_all().await?;
let index = RowIdFileIndex::from_splits(scan_table.clone(),
plan.splits())?;
let source = match source {
MergeSource::Batches(batches) => Some(MergeInput::new(batches)?),
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 12fbfe0d..119e88eb 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -66,6 +66,9 @@ pub use audit::AuditLogScan;
#[path = "table_scan/restore_splits.rs"]
mod restore_splits;
+mod file_scan;
+pub use file_scan::FileScanPlan;
+
/// Path segment for manifest directory under table.
const MANIFEST_DIR: &str = "manifest";
/// Path segment for index directory under table.
@@ -1127,6 +1130,7 @@ impl<'a> TableScan<'a> {
split_selection: scan.split_selection,
row_range_optimization_disabled:
scan.row_range_optimization_disabled,
scan_all_files: scan.scan_all_files,
+ file_scan_options: scan.file_scan_options,
incremental_split_mode: scan.incremental_split_mode,
projected_read_field_ids: scan.projected_read_field_ids,
query_auth_grant: scan.query_auth_grant,
@@ -1513,10 +1517,10 @@ impl<'a> TableScan<'a> {
}
}
- /// Plan matching files including L0 and all data-evolution column files.
+ /// Plan splits including L0 and all matching data-evolution column files.
/// Existing scan filters and limits still apply. Query authorization must
/// be unrestricted, as required by writer restore, overwrite and indexing.
- pub async fn plan_all_files(&self) -> crate::Result<Plan> {
+ pub async fn plan_all(&self) -> crate::Result<Plan> {
let mut source = self.clone();
if let TableScanKind::Paimon(scan) = &mut source.0 {
scan.projected_read_field_ids = None;
@@ -1729,6 +1733,7 @@ struct PaimonTableScan<'a> {
/// Used by non-read paths (overwrite, truncate, writer restore) that need
/// the complete file set. Normal read scans leave this as `false`.
scan_all_files: bool,
+ file_scan_options: Option<file_scan::FileScanOptions>,
incremental_split_mode: Option<IncrementalSplitMode>,
projected_read_field_ids: Option<HashSet<i32>>,
/// Only set on an ephemeral copy after authorization. Never stored back
into
@@ -1796,6 +1801,9 @@ impl<'a> PaimonTableScan<'a> {
}
fn only_real_buckets(&self) -> bool {
+ if let Some(options) = self.file_scan_options {
+ return options.only_read_real_buckets;
+ }
self.split_selection
.as_deref()
.is_some_and(|selection| selection.only_real_buckets)
@@ -1857,6 +1865,7 @@ impl<'a> PaimonTableScan<'a> {
split_selection: None,
row_range_optimization_disabled: false,
scan_all_files: false,
+ file_scan_options: None,
incremental_split_mode: None,
projected_read_field_ids: None,
query_auth_grant: None,
@@ -1869,6 +1878,11 @@ impl<'a> PaimonTableScan<'a> {
/// the complete file set regardless of merge engine or DV settings.
pub fn with_scan_all_files(mut self) -> Self {
self.scan_all_files = true;
+ if let Some(options) = self.file_scan_options.take() {
+ self.split_selection
+ .get_or_insert_with(Default::default)
+ .only_real_buckets = options.only_read_real_buckets;
+ }
self.projected_read_field_ids = None;
self
}
@@ -2096,8 +2110,8 @@ impl<'a> PaimonTableScan<'a> {
/// The grant predates the manifest read, so reject a concurrent
replacement
/// or rule-schema change. File metadata, including old column statistics,
/// is carried unchanged by Java QueryAuthSplit as well.
- async fn check_planned_files(&self, _plan: &Plan, query_auth: bool) ->
crate::Result<()> {
- if query_auth {
+ async fn check_planned_files(&self) -> crate::Result<()> {
+ if self.query_auth_grant.is_some() {
if let Some(rest_env) = self.table.rest_env() {
if let Some((id, fields)) = self
.query_auth_grant
@@ -2143,8 +2157,7 @@ impl<'a> PaimonTableScan<'a> {
}
async fn finish_plan(&self, plan: Plan) -> crate::Result<Plan> {
- self.check_planned_files(&plan, self.query_auth_grant.is_some())
- .await?;
+ self.check_planned_files().await?;
Ok(plan.planned(self.query_auth_grant.clone()))
}
@@ -2255,12 +2268,19 @@ impl<'a> PaimonTableScan<'a> {
let options = self.table.schema().core_options();
let has_primary_keys = !self.table.schema().primary_keys().is_empty();
let initial_state = self.snapshot_mode() ==
Some(SnapshotScanMode::All);
- // Initial stream state merges overlapping key versions, including
- // L0 files. Keep each bucket intact until that merge has completed.
+ let includes_level_zero = self
+ .file_scan_options
+ .is_some_and(|options| !options.skip_level0);
+ let includes_pending = self
+ .file_scan_options
+ .is_some_and(|options| !options.only_read_real_buckets);
+ // Initial stream state and explicit L0 reads merge overlapping key
+ // versions. Keep each bucket intact until that merge has completed.
// Follow-up physical events and ordinary raw scans retain Java's
// SplitGenerator.alwaysRawConvertible file-name distribution.
let by_file_name = if has_primary_keys {
!initial_state
+ && !includes_level_zero
&& (options.deletion_vectors_enabled()
|| options.merge_engine()? ==
crate::spec::MergeEngine::FirstRow)
} else {
@@ -2280,9 +2300,13 @@ impl<'a> PaimonTableScan<'a> {
((i128::from(hash) % count as i128).unsigned_abs() as usize)
== index
} else {
let bucket = i128::from(entry.bucket());
- // Initial state includes postpone buckets. Assign them to
- // one worker as well, rather than dropping a negative
remainder.
- let bucket = if initial_state { bucket.abs() } else { bucket };
+ // Initial state and explicit all-bucket reads include postpone
+ // buckets. Assign them to one worker as well.
+ let bucket = if initial_state || includes_pending {
+ bucket.abs()
+ } else {
+ bucket
+ };
bucket % count as i128 == index as i128
}
});
@@ -2406,7 +2430,10 @@ impl<'a> PaimonTableScan<'a> {
&self.table,
source,
skip_level_zero,
- self.scan_all_files,
+ self.file_scan_options
+ .map_or(self.scan_all_files, |options| {
+ !options.only_read_real_buckets
+ }),
has_primary_keys,
self.partition_filter.as_ref(),
&partition_fields,
@@ -2615,6 +2642,9 @@ impl<'a> PaimonTableScan<'a> {
let options = self.table.schema().core_options();
self.snapshot_mode() == Some(SnapshotScanMode::All)
|| self.is_streaming()
+ || self
+ .file_scan_options
+ .is_some_and(|options| !options.skip_level0)
|| match options.merge_engine() {
Ok(crate::spec::MergeEngine::FirstRow) =>
!self.skip_level_zero(),
_ => {
@@ -2624,6 +2654,9 @@ impl<'a> PaimonTableScan<'a> {
}
fn skip_level_zero(&self) -> bool {
+ if let Some(options) = self.file_scan_options {
+ return options.skip_level0;
+ }
let options = self.table.schema().core_options();
should_skip_level_zero_for_scan(
self.scan_all_files || self.is_streaming(),
@@ -7780,6 +7813,192 @@ mod tests {
assert!(empty.splits().is_empty());
assert_eq!(empty.snapshot_id(), Some(1));
}
+ #[tokio::test]
+ async fn test_file_scan_merges_files_before_limits_and_distribution() {
+ let table = scan_stats_test_table("memory:/file_scan_merged");
+ setup_scan_stats_dirs(&table).await;
+ let partition = BinaryRowBuilder::new(0).build_serialized();
+ let removed = stats_test_file("removed.parquet", 1, 2);
+ let keep = stats_test_file("keep.parquet", 10, 20);
+ let commit = TableCommit::new(table.clone(), "file-scan".into());
+ commit
+ .commit(vec![CommitMessage::new(
+ partition.clone(),
+ 0,
+ vec![removed.clone(), keep.clone()],
+ )])
+ .await
+ .unwrap();
+ let mut changes =
+ CommitMessage::new(partition, 0,
vec![stats_test_file("new.parquet", 100, 101)]);
+ changes.deleted_files.push(removed);
+ commit.commit(vec![changes]).await.unwrap();
+
+ let mut builder = table.new_read_builder();
+ builder.with_limit(0);
+ builder.with_row_ranges(Vec::new());
+ let plan = builder
+ .new_scan()
+ .with_row_position_slice(0, 1)
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ assert_eq!(
+ plan.files()
+ .iter()
+ .map(|e| e.file().file_name.as_str())
+ .collect::<Vec<_>>(),
+ ["keep.parquet", "new.parquet"]
+ );
+ let sharded = builder
+ .new_scan()
+ .with_shard(1, 2)
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap();
+ assert_eq!(sharded.files(), plan.files());
+ builder.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .equal("id", Datum::Int(15))
+ .unwrap(),
+ );
+ let filtered = builder.new_scan().plan_files().await.unwrap();
+ assert_eq!(filtered.files().len(), 1);
+ assert_eq!(filtered.files()[0].file().file_name, "keep.parquet");
+ assert_eq!(filtered.files()[0].file().value_stats, keep.value_stats);
+
+ let historical =
+ table.copy_with_options(HashMap::from([("scan.snapshot-id".into(),
"1".into())]));
+ let old = historical
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .unwrap();
+ assert_eq!(old.snapshot_id(), Some(1));
+ assert_eq!(
+ old.files()
+ .iter()
+ .map(|e| e.file().file_name.as_str())
+ .collect::<Vec<_>>(),
+ ["removed.parquet", "keep.parquet"]
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_file_scan_level_and_pending_bucket_policies_are_independent() {
+ let schema = PaimonSchema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "-2")
+ .option("deletion-vectors.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "policies"),
+ "memory:/file_scan_policies".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ setup_scan_stats_dirs(&table).await;
+ let file = |name: &str, level| {
+ let mut file = make_evo_file(name, 128, 2, 1, None);
+ file.level = level;
+ file
+ };
+ let mut real = CommitMessage::new(
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ vec![file("real-l0.parquet", 0), file("real-l1.parquet", 1)],
+ );
+ real.total_buckets = Some(2);
+ let pending = CommitMessage::new(
+ BinaryRowBuilder::new(0).build_serialized(),
+ -2,
+ vec![file("pending-l0.parquet", 0), file("pending-l1.parquet", 1)],
+ );
+ TableCommit::new(table.clone(), "file-policies".into())
+ .commit(vec![real, pending])
+ .await
+ .unwrap();
+ let builder = table.new_read_builder();
+ let mut scan = builder.new_scan();
+ assert_eq!(scan.plan_files().await.unwrap().files().len(), 4);
+ for (skip, only_real, expected) in [
+ (false, false, 4),
+ (true, false, 2),
+ (false, true, 2),
+ (true, true, 1),
+ (false, false, 4),
+ ] {
+ scan = scan.with_file_filters(skip, only_real).unwrap();
+ let plan = scan.plan_files().await.unwrap();
+ assert_eq!(plan.files().len(), expected);
+ assert!(plan
+ .files()
+ .iter()
+ .all(|e| (!skip || e.file().level != 0) && (!only_real ||
e.bucket() >= 0)));
+ }
+ // Explicit metadata policies must not change a later query scan's
defaults.
+ let query = table.new_read_builder().new_scan().plan().await.unwrap();
+ assert!(query
+ .splits()
+ .iter()
+ .all(|s| s.bucket() >= 0 && s.data_files().iter().all(|f| f.level
> 0)));
+ }
+
+ #[tokio::test]
+ async fn
test_file_scan_keeps_unprojected_blob_files_and_ignores_index_ranges() {
+ let schema = PaimonSchema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(crate::spec::BlobType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = data_evolution_test_table(
+ "memory:/file_scan_blob_columns",
+ TableSchema::new(0, &schema),
+ );
+ setup_scan_stats_dirs(&table).await;
+ let mut anchor = make_evo_file_with_cols("anchor.parquet", 4, 1, 0,
&["id"]);
+ anchor.first_row_id = None;
+ anchor.file_source = Some(0);
+ let mut blob = make_evo_file_with_cols("payload.blob", 4, 1, 0,
&["payload"]);
+ blob.file_source = Some(0);
+ let commit = TableCommit::new(table.clone(), "file-columns".into());
+ commit
+ .commit(vec![CommitMessage::new(
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ vec![anchor],
+ )])
+ .await
+ .unwrap();
+ commit
+ .commit(vec![CommitMessage::new(
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ vec![blob],
+ )])
+ .await
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["id"]).unwrap().with_limit(0);
+ let plan = builder
+ .new_scan()
+
.with_global_index_result(super::GlobalIndexResult::from_row_ranges(Vec::new()))
+ .unwrap()
+ .plan_files()
+ .await
+ .unwrap();
+ assert_eq!(plan.files().len(), 2);
+ assert_eq!(plan.files()[1].file().file_name, "payload.blob");
+ }
}
#[cfg(test)]
diff --git a/crates/paimon/src/table/table_scan/file_scan.rs
b/crates/paimon/src/table/table_scan/file_scan.rs
new file mode 100644
index 00000000..93dfab5b
--- /dev/null
+++ b/crates/paimon/src/table/table_scan/file_scan.rs
@@ -0,0 +1,234 @@
+// 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.
+
+//! File metadata planning before row selection and split packing.
+
+use super::{
+ read_all_manifest_entries, BucketSelector, CoreOptions, ManifestEntry,
ManifestListSource,
+ PaimonTableScan, ScanStats, TableScan, TableScanKind,
+};
+use crate::table::incremental_scan::timestamp_snapshot_range;
+use crate::table::{IncrementalScan, IncrementalScanMode};
+use std::collections::HashMap;
+
+#[derive(Debug, Clone, Copy)]
+pub(super) struct FileScanOptions {
+ pub(super) skip_level0: bool,
+ pub(super) only_read_real_buckets: bool,
+}
+
+/// A selected snapshot's filtered files, before split packing or row
selection.
+#[derive(Debug, Clone)]
+pub struct FileScanPlan {
+ files: Vec<ManifestEntry>,
+ snapshot_id: Option<i64>,
+}
+
+impl FileScanPlan {
+ pub fn files(&self) -> &[ManifestEntry] {
+ &self.files
+ }
+
+ /// None for an empty table or a timestamp window with no selected
snapshot.
+ pub fn snapshot_id(&self) -> Option<i64> {
+ self.snapshot_id
+ }
+}
+
+impl TableScan<'_> {
+ /// Override file-level policies used by [`Self::plan`] and
[`Self::plan_files`].
+ /// Snapshot selection and whole-file metadata authorization are unchanged.
+ /// Unconfigured queries keep their normal merge-engine and deletion-vector
+ /// policies. [`Self::plan_all`] disables the level-zero exclusion while
+ /// preserving the configured bucket policy.
+ pub fn with_file_filters(
+ mut self,
+ skip_level0: bool,
+ only_read_real_buckets: bool,
+ ) -> crate::Result<Self> {
+ let TableScanKind::Paimon(scan) = &mut self.0 else {
+ return Err(crate::Error::Unsupported {
+ message: "File filters require a Paimon table".into(),
+ });
+ };
+ scan.file_scan_options = Some(FileScanOptions {
+ skip_level0,
+ only_read_real_buckets,
+ });
+ Ok(self)
+ }
+
+ /// Plan filtered, merged file entries without applying projections, LIMIT,
+ /// row ranges, split distribution, or global-index selections.
+ ///
+ /// The independent policies match FileStoreScan: skipping level zero never
+ /// changes whether pending buckets are visible. Metadata requires an
+ /// unrestricted whole-table grant even when the row reader projects
fields.
+ /// By default include level zero and pending buckets; configured bucket
+ /// exclusions also survive [`Self::with_scan_all_files`].
+ pub async fn plan_files(&self) -> crate::Result<FileScanPlan> {
+ let mut requested = self.clone();
+ let TableScanKind::Paimon(scan) = &mut requested.0 else {
+ return Err(crate::Error::Unsupported {
+ message: "Format tables do not have Paimon manifest
entries".into(),
+ });
+ };
+ let filters = scan.file_scan_options.unwrap_or(FileScanOptions {
+ skip_level0: false,
+ only_read_real_buckets: scan.only_real_buckets(),
+ });
+ scan.projected_read_field_ids = None;
+ scan.row_ranges = None;
+ scan.split_selection = None;
+ // Preserve the effective raw-file policy for the shared merge-safe
stats
+ // rules. Raw defaults include L0 even when ordinary first-row reads
do not.
+ scan.file_scan_options = Some(filters);
+ scan.limit = None;
+ // Authorize before turning on trusted metadata-reader behavior.
+ scan.scan_all_files = false;
+ let authorized = requested.authorize_all_files().await?;
+ let TableScanKind::Paimon(scan) = &authorized.0 else {
+ unreachable!("file metadata planning only accepts Paimon tables")
+ };
+ let plan = scan
+ .plan_file_entries(filters.skip_level0,
filters.only_read_real_buckets)
+ .await?;
+ scan.check_planned_files().await?;
+ Ok(plan)
+ }
+}
+
+impl PaimonTableScan<'_> {
+ async fn plan_file_entries(
+ &self,
+ skip_level0: bool,
+ only_read_real_buckets: bool,
+ ) -> crate::Result<FileScanPlan> {
+ let options = self.table.schema().core_options();
+ if let Some((start, end)) = options.incremental_timestamp_window()? {
+ let Some((start, end)) =
+ timestamp_snapshot_range(&self.table.snapshot_manager(),
start, end).await?
+ else {
+ return Ok(FileScanPlan {
+ files: Vec::new(),
+ snapshot_id: None,
+ });
+ };
+ let incremental = IncrementalScan::new(
+ &self.table,
+ TableScan(TableScanKind::Paimon(self.clone())),
+ IncrementalScanMode::Auto,
+ start,
+ end,
+ );
+ let (snapshot, lists) =
incremental.selected_event_manifest_lists().await?;
+ let files = self
+ .read_file_entries(
+ ManifestListSource::IncrementalLists(lists),
+ skip_level0,
+ only_read_real_buckets,
+ )
+ .await?;
+ return Ok(FileScanPlan {
+ files,
+ snapshot_id: Some(snapshot.id()),
+ });
+ }
+ let Some(snapshot) =
super::super::time_travel::resolve_snapshot(&self.table).await? else {
+ return Ok(FileScanPlan {
+ files: Vec::new(),
+ snapshot_id: None,
+ });
+ };
+ let files = self
+ .read_file_entries(
+ ManifestListSource::Snapshot(&snapshot),
+ skip_level0,
+ only_read_real_buckets,
+ )
+ .await?;
+ Ok(FileScanPlan {
+ files,
+ snapshot_id: Some(snapshot.id()),
+ })
+ }
+
+ async fn read_file_entries(
+ &self,
+ source: ManifestListSource<'_>,
+ skip_level0: bool,
+ only_read_real_buckets: bool,
+ ) -> crate::Result<Vec<ManifestEntry>> {
+ let options = CoreOptions::new(self.table.schema().options());
+ let primary_key = !self.table.schema().primary_keys().is_empty();
+ let data_predicates = if options.data_evolution_enabled() {
+ Vec::new()
+ } else {
+ self.stats_pruning_predicates()
+ };
+ let key_predicates = self.key_stats_predicates(&data_predicates);
+ let bucket_selector = self
+ .bucket_predicate
+ .as_ref()
+ .map(|predicate| BucketSelector::new(&self.table, predicate))
+ .transpose()?
+ .flatten();
+ let entries = read_all_manifest_entries(
+ &self.table,
+ source,
+ skip_level0,
+ true,
+ primary_key,
+ self.partition_filter.as_ref(),
+ &self.table.schema().partition_fields(),
+ &data_predicates,
+ &key_predicates,
+ self.table.schema().id(),
+ self.table.schema().fields(),
+ &self.table.schema().trimmed_primary_keys(),
+ bucket_selector.as_ref(),
+ None,
+ options.manifest_sidecar_enabled(),
+ options.scan_manifest_parallelism()?,
+ &mut ScanStats::default(),
+ )
+ .await?;
+ let mut files = Vec::with_capacity(entries.len());
+ let mut schema_cache = HashMap::new();
+ for entry in entries {
+ if only_read_real_buckets && entry.bucket() < 0 {
+ continue;
+ }
+ // Older-schema stats need the same field-id evolution as row
scans.
+ if entry.file().schema_id != self.table.schema().id()
+ && !data_predicates.is_empty()
+ && !super::data_file_matches_predicates_for_table(
+ &self.table,
+ entry.file(),
+ &data_predicates,
+ &key_predicates,
+ &mut schema_cache,
+ )
+ .await
+ {
+ continue;
+ }
+ files.push(entry);
+ }
+ Ok(files)
+ }
+}
diff --git a/crates/paimon/src/table/table_upsert.rs
b/crates/paimon/src/table/table_upsert.rs
index fc082a02..73d62647 100644
--- a/crates/paimon/src/table/table_upsert.rs
+++ b/crates/paimon/src/table/table_upsert.rs
@@ -400,7 +400,7 @@ impl TableUpsert {
let mut projection =
self.keys.iter().map(String::as_str).collect::<Vec<_>>();
projection.push(ROW_ID);
read_builder.with_projection(&projection)?;
- let plan = read_builder.new_scan().plan_all_files().await?;
+ let plan = read_builder.new_scan().plan_all().await?;
let read = read_builder.new_read()?.for_update();
let splits = self.matching_splits(&source, &plan)?;
let mut stream = read.to_arrow(splits.as_ref())?;
diff --git a/crates/paimon/tests/incremental_timestamp_scan_test.rs
b/crates/paimon/tests/incremental_timestamp_scan_test.rs
index 9f0da25d..090614b4 100644
--- a/crates/paimon/tests/incremental_timestamp_scan_test.rs
+++ b/crates/paimon/tests/incremental_timestamp_scan_test.rs
@@ -158,6 +158,30 @@ async fn
timestamp_options_select_events_and_preserve_end_snapshot() {
assert_eq!(rows, expected, "{mode}: ({start},{end}]");
}
let plan =
selected.new_read_builder().new_scan().plan().await.unwrap();
+ let files = selected
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .unwrap();
+ assert_eq!(files.snapshot_id(), snapshot);
+ let mut actual_names = files
+ .files()
+ .iter()
+ .map(|entry| entry.file().file_name.clone())
+ .collect::<Vec<_>>();
+ let mut expected_names = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .map(|file| file.file_name.clone())
+ .collect::<Vec<_>>();
+ actual_names.sort();
+ expected_names.sort();
+ assert_eq!(
+ actual_names, expected_names,
+ "file events {mode}: ({start},{end}]"
+ );
let scan_stats = plan.stats().clone();
assert_eq!(plan.snapshot_id(), snapshot);
assert_eq!(scan_stats.final_splits, plan.splits().len());
@@ -183,6 +207,23 @@ async fn
timestamp_auto_uses_physical_changelog_and_preserves_retracts() {
let selected = window(&table, 100, 300);
let builder = selected.new_read_builder();
let plan = builder.new_scan().plan().await.unwrap();
+ let files = builder.new_scan().plan_files().await.unwrap();
+ assert_eq!(files.snapshot_id(), Some(3));
+ let actual_names = files
+ .files()
+ .iter()
+ .map(|entry| entry.file().file_name.as_str())
+ .collect::<Vec<_>>();
+ let expected_names = plan
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .map(|file| file.file_name.as_str())
+ .collect::<Vec<_>>();
+ assert_eq!(
+ actual_names, expected_names,
+ "AUTO must select physical changelog files"
+ );
let scan_stats = plan.stats().clone();
assert_eq!(plan.snapshot_id(), Some(3));
assert!(plan.splits().iter().all(|split| split.snapshot_id() == 3));
diff --git a/crates/paimon/tests/nullable_primary_key_test.rs
b/crates/paimon/tests/nullable_primary_key_test.rs
index 9ca5ed1b..c56e0988 100644
--- a/crates/paimon/tests/nullable_primary_key_test.rs
+++ b/crates/paimon/tests/nullable_primary_key_test.rs
@@ -82,7 +82,7 @@ async fn read(table: &Table, predicate: Option<Predicate>) ->
Vec<Row> {
if let Some(predicate) = predicate {
builder.with_filter(predicate);
}
- // First-row source reads include L0, as Python plan_all_files does.
+ // First-row source reads include L0, as Python plan_all does.
let plan = builder
.new_scan()
.with_scan_all_files()
diff --git a/crates/paimon/tests/rest_catalog/merge_sources.rs
b/crates/paimon/tests/rest_catalog/merge_sources.rs
index 3f2917c5..756d4aa2 100644
--- a/crates/paimon/tests/rest_catalog/merge_sources.rs
+++ b/crates/paimon/tests/rest_catalog/merge_sources.rs
@@ -423,7 +423,7 @@ async fn
merge_rejects_restricted_sources_even_for_unused_fields_or_empty_views(
source.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 1)], &[]);
}
// DELETE matches only `id`: an unused column's mask still makes
the
- // source unsuitable for Python's existing plan_all_files contract.
+ // source unsuitable for Python's existing plan_all contract.
let error = merge(&target, &source.table, true).await.unwrap_err();
assert!(error.to_string().contains("unrestricted"), "{error}");
assert_eq!(
diff --git a/crates/paimon/tests/rest_catalog/read_path_completion.rs
b/crates/paimon/tests/rest_catalog/read_path_completion.rs
index e8d7eb98..97fa39c0 100644
--- a/crates/paimon/tests/rest_catalog/read_path_completion.rs
+++ b/crates/paimon/tests/rest_catalog/read_path_completion.rs
@@ -176,7 +176,7 @@ async fn
write_planning_rejects_rules_and_keeps_all_file_scan_guard() {
.table
.new_read_builder()
.new_scan()
- .plan_all_files()
+ .plan_all()
.await
.unwrap()
.splits()
@@ -186,7 +186,7 @@ async fn
write_planning_rejects_rules_and_keeps_all_file_scan_guard() {
.set_column_auth("default", "write_plan", vec!["id".into()]);
let mut projected = g.table.new_read_builder();
projected.with_projection(&["id"]).unwrap();
- assert_forbidden(projected.new_scan().plan_all_files().await.unwrap_err());
+ assert_forbidden(projected.new_scan().plan_all().await.unwrap_err());
g.ctx
.server
.set_column_auth("default", "write_plan", vec!["id".into(),
"name".into()]);
@@ -195,7 +195,7 @@ async fn
write_planning_rejects_rules_and_keeps_all_file_scan_guard() {
.table
.new_read_builder()
.new_scan()
- .plan_all_files()
+ .plan_all()
.await
.is_err());
assert!(g
@@ -864,3 +864,115 @@ async fn
transported_optional_bucket_layout_uses_canonical_snapshot_paths() {
[1, 2]
);
}
+
+#[tokio::test]
+async fn file_scan_requires_unrestricted_whole_table_authorization() {
+ let g = written("raw_files", people, vec![people_rows(1..=2)]).await;
+ let mut projected = g.table.new_read_builder();
+ projected.with_projection(&["id"]).unwrap().with_limit(0);
+ let files = projected.new_scan().plan_files().await.unwrap();
+ assert_eq!(files.snapshot_id(), Some(1));
+ assert!(
+ !files.files().is_empty(),
+ "row LIMIT must not hide metadata"
+ );
+ assert_eq!(
+ g.ctx.server.auth_selects("default", "raw_files").last(),
+ Some(&None)
+ );
+
+ g.ctx
+ .server
+ .set_column_auth("default", "raw_files", vec!["id".into()]);
+ assert_forbidden(projected.new_scan().plan_files().await.unwrap_err());
+ g.ctx
+ .server
+ .set_column_auth("default", "raw_files", vec!["id".into(),
"name".into()]);
+ for (filters, masks) in [
+ (vec![int_leaf(0, "id", "GREATER_THAN", 1)], vec![]),
+ (vec![], vec![("name", upper(1, "name"))]),
+ ] {
+ g.set_rules(&filters, &masks);
+ assert!(projected.new_scan().plan_files().await.is_err());
+ let empty = g.table.copy_with_options(HashMap::from([
+ ("incremental-between-timestamp".into(), "100,100".into()),
+ ("scan.mode".into(), "incremental".into()),
+ ]));
+ assert!(
+ empty
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .is_err(),
+ "empty timestamp windows must authorize before returning metadata"
+ );
+ }
+}
+
+#[tokio::test]
+async fn file_scan_refuses_a_replaced_schema() {
+ let g = guarded("file_schema", &["id"]).await;
+ let replaced = g
+ .table
+ .copy_with_resolved_schema(
+ paimon::spec::TableSchema::new(0, &schema_of(&["id", "secret"],
GUARDED)),
+ "main",
+ )
+ .unwrap();
+ assert_refused(
+ replaced
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .unwrap_err(),
+ );
+ g.ctx.server.set_table_schema_id(
+ "default",
+ "file_schema",
+ schema_of(&["id", "extra"], GUARDED),
+ 0,
+ );
+ assert_drifted(
+ g.table
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .unwrap_err(),
+ "serves other columns",
+ );
+}
+
+#[tokio::test]
+async fn file_scan_checks_identity_after_planning_empty_file_sets() {
+ for timestamp_window in [false, true] {
+ let name = if timestamp_window {
+ "empty_file_window"
+ } else {
+ "empty_file_table"
+ };
+ let g = guarded(name, &["id"]).await;
+ let selected = if timestamp_window {
+ g.table.copy_with_options(HashMap::from([
+ ("incremental-between-timestamp".into(), "100,100".into()),
+ ("scan.mode".into(), "incremental".into()),
+ ]))
+ } else {
+ g.table.clone()
+ };
+ g.ctx
+ .server
+ .set_table_uuid_after_calls("default", name,
"replacement-after-grant", 2);
+ assert_drifted(
+ selected
+ .new_read_builder()
+ .new_scan()
+ .plan_files()
+ .await
+ .unwrap_err(),
+ "now resolves to uuid",
+ );
+ }
+}
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 164f76b2..3bb470f7 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -4964,7 +4964,7 @@ async fn
test_query_auth_keeps_a_mask_source_in_another_partial_file() {
.sum::<usize>(),
1
);
- let write_plan = projected.new_scan().plan_all_files().await.unwrap();
+ let write_plan = projected.new_scan().plan_all().await.unwrap();
assert_eq!(
write_plan
.splits()
