[core] Support authorized raw file planning for native scans

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."""
 

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]

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>()?;

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,18 +795,43 @@ 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)
             })
         })
     }
 
+    /// 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| {