diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 81770d157..ad573e03c 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 7bb744c31..0dbe0a713 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 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 102a3a4ae..7e56cedaf 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::()?; this.add_class::()?; this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs index 30ffdf528..05996167d 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, 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> { + 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 { + /// 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 { 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 { + 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 { 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 { + self.inner.snapshot_id() + } + + fn files(&self) -> Vec { + self.inner.files().iter().cloned().map(Into::into).collect() + } +} + impl From 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::(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::() + .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::().unwrap(), 0); + assert_eq!( + first + .getattr("total_buckets") + .unwrap() + .extract::() + .unwrap(), + 1 + ); + let file = first.getattr("file").unwrap(); + assert_eq!( + file.getattr("row_count").unwrap().extract::().unwrap(), + 1 + ); + assert!(file + .getattr("file_name") + .unwrap() + .extract::() + .unwrap() + .ends_with(".parquet")); + let encoded = first + .call_method0("_to_commit_message") + .unwrap() + .call_method0("serialize") + .unwrap(); + let encoded = encoded.cast::().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::>().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 aa9d8cb65..eb24ea093 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 38c5aaa41..2df6a7855 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 70ece9224..d06b4c03b 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)> { + 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 7cbce5ba2..0fee5af87 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 3872af02e..36695190d 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 c535dc2a7..74fe00a43 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 8c95ee071..ffdc71c64 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 4aa129dbe..4abdff749 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::>())?; // 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 12fbfe0de..119e88eb6 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 { + pub async fn plan_all(&self) -> crate::Result { 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, incremental_split_mode: Option, projected_read_field_ids: Option>, /// 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 { - 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::>(), + ["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::>(), + ["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 000000000..93dfab5b4 --- /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, + snapshot_id: Option, +} + +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 { + 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 { + 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 { + 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 { + 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> { + 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 fc082a025..73d626470 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::>(); 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 9f0da25d3..090614b49 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::>(); + let mut expected_names = plan + .splits() + .iter() + .flat_map(|split| split.data_files()) + .map(|file| file.file_name.clone()) + .collect::>(); + 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::>(); + let expected_names = plan + .splits() + .iter() + .flat_map(|split| split.data_files()) + .map(|file| file.file_name.as_str()) + .collect::>(); + 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 9ca5ed1b0..c56e09885 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) -> Vec { 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 3f2917c5c..756d4aa2b 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 e8d7eb98b..97fa39c04 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 164f76b2c..3bb470f78 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::(), 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()