[core] Improve fixed-bucket pruning and expose scan statistics

bindings/python/README.md

@@ -158,6 +158,22 @@ print(f"\nRead: {batches_tt[0].num_rows} rows")
 print(batches_tt[0])
 ```
 
+### Scan statistics
+
+Every scan collects lightweight counters during ordinary metadata planning:
+
+```python
+plan = table.new_read_builder().new_scan().plan()
+scan_stats = plan.stats()
+print(scan_stats.manifest_entries_read, scan_stats.final_files)
+```
+
+Collection retains early manifest/sidecar pruning and LIMIT stopping. Counters
+record the work performed; skipped entries are not decoded to reconstruct totals.
+Combined incremental scans expose
+the same API for their selected Delta or Changelog window. Snapshot metadata is
+available from `plan.snapshot_id()`.
+
 ### Shard column updates
 
 The batch and stream update objects expose

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -65,8 +65,68 @@ class Plan:
         """Selected snapshot, preserved for empty plans; None if no snapshot was selected."""
         ...
     def splits(self) -> List[Split]: ...
+    def stats(self) -> ScanStats:
+        """Lightweight statistics collected during ordinary planning."""
+        ...
     def __len__(self) -> int: ...
 
+class ScanStats:
+    """Read-only counters from normal metadata planning."""
+    @property
+    def base_manifest_files(self) -> int: ...
+    @property
+    def delta_manifest_files(self) -> int: ...
+    @property
+    def manifest_files_before_partition_pruning(self) -> int: ...
+    @property
+    def manifest_files_after_partition_pruning(self) -> int: ...
+    @property
+    def manifest_files_pruned_by_row_ranges(self) -> int: ...
+    @property
+    def manifest_entries_read(self) -> int: ...
+    @property
+    def manifest_entries_pruned_by_bucket(self) -> int: ...
+    @property
+    def manifest_entries_pruned_by_partition(self) -> int: ...
+    @property
+    def manifest_entries_after_entry_pruning(self) -> int: ...
+    @property
+    def manifest_entries_pruned_by_level(self) -> int: ...
+    @property
+    def manifest_entries_pruned_by_row_ranges(self) -> int: ...
+    @property
+    def manifest_entries_pruned_by_data_stats(self) -> int: ...
+    @property
+    def manifest_entries_after_manifest_filters(self) -> int: ...
+    @property
+    def manifest_entries_after_merge(self) -> int: ...
+    @property
+    def manifest_entries_pruned_by_cross_schema_stats(self) -> int: ...
+    @property
+    def manifest_entries_after_cross_schema_stats(self) -> int: ...
+    @property
+    def data_evolution_groups_before_stats(self) -> int: ...
+    @property
+    def data_evolution_groups_pruned_by_stats(self) -> int: ...
+    @property
+    def data_evolution_groups_pruned_by_row_ranges(self) -> int: ...
+    @property
+    def split_candidates_built(self) -> int: ...
+    @property
+    def limit_early_stopped(self) -> bool: ...
+    @property
+    def splits_before_limit(self) -> int: ...
+    @property
+    def splits_after_limit(self) -> int: ...
+    @property
+    def final_splits(self) -> int: ...
+    @property
+    def final_files(self) -> int: ...
+    @property
+    def planned_data_file_bytes(self) -> int: ...
+    @property
+    def limit(self) -> Optional[int]: ...
+
 class VindexVectorGlobalIndexReader:
     """Rust index search over an input exposing pread_many(ranges)."""
     def __init__(

bindings/python/src/context.rs

@@ -817,6 +817,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::PyScanStats>()?;
     this.add_class::<crate::read::PyTableRead>()?;
     this.add_class::<crate::read::PyRecordBatchReader>()?;
     this.add_class::<crate::read::PySplit>()?;

bindings/python/src/read.rs

@@ -758,6 +758,123 @@ impl PyTableScan {
     }
 }
 
+#[pyclass(name = "ScanStats", module = "pypaimon_rust.datafusion", frozen)]
+pub struct PyScanStats {
+    inner: paimon::table::ScanStats,
+}
+
+#[pymethods]
+impl PyScanStats {
+    #[getter]
+    fn base_manifest_files(&self) -> usize {
+        self.inner.base_manifest_files
+    }
+    #[getter]
+    fn delta_manifest_files(&self) -> usize {
+        self.inner.delta_manifest_files
+    }
+    #[getter]
+    fn manifest_files_before_partition_pruning(&self) -> usize {
+        self.inner.manifest_files_before_partition_pruning
+    }
+    #[getter]
+    fn manifest_files_after_partition_pruning(&self) -> usize {
+        self.inner.manifest_files_after_partition_pruning
+    }
+    #[getter]
+    fn manifest_files_pruned_by_row_ranges(&self) -> usize {
+        self.inner.manifest_files_pruned_by_row_ranges
+    }
+    #[getter]
+    fn manifest_entries_read(&self) -> usize {
+        self.inner.manifest_entries_read
+    }
+    #[getter]
+    fn manifest_entries_pruned_by_bucket(&self) -> usize {
+        self.inner.manifest_entries_pruned_by_bucket
+    }
+    #[getter]
+    fn manifest_entries_pruned_by_partition(&self) -> usize {
+        self.inner.manifest_entries_pruned_by_partition
+    }
+    #[getter]
+    fn manifest_entries_after_entry_pruning(&self) -> usize {
+        self.inner.manifest_entries_after_entry_pruning
+    }
+    #[getter]
+    fn manifest_entries_pruned_by_level(&self) -> usize {
+        self.inner.manifest_entries_pruned_by_level
+    }
+    #[getter]
+    fn manifest_entries_pruned_by_row_ranges(&self) -> usize {
+        self.inner.manifest_entries_pruned_by_row_ranges
+    }
+    #[getter]
+    fn manifest_entries_pruned_by_data_stats(&self) -> usize {
+        self.inner.manifest_entries_pruned_by_data_stats
+    }
+    #[getter]
+    fn manifest_entries_after_manifest_filters(&self) -> usize {
+        self.inner.manifest_entries_after_manifest_filters
+    }
+    #[getter]
+    fn manifest_entries_after_merge(&self) -> usize {
+        self.inner.manifest_entries_after_merge
+    }
+    #[getter]
+    fn manifest_entries_pruned_by_cross_schema_stats(&self) -> usize {
+        self.inner.manifest_entries_pruned_by_cross_schema_stats
+    }
+    #[getter]
+    fn manifest_entries_after_cross_schema_stats(&self) -> usize {
+        self.inner.manifest_entries_after_cross_schema_stats
+    }
+    #[getter]
+    fn data_evolution_groups_before_stats(&self) -> usize {
+        self.inner.data_evolution_groups_before_stats
+    }
+    #[getter]
+    fn data_evolution_groups_pruned_by_stats(&self) -> usize {
+        self.inner.data_evolution_groups_pruned_by_stats
+    }
+    #[getter]
+    fn data_evolution_groups_pruned_by_row_ranges(&self) -> usize {
+        self.inner.data_evolution_groups_pruned_by_row_ranges
+    }
+    #[getter]
+    fn split_candidates_built(&self) -> usize {
+        self.inner.split_candidates_built
+    }
+    #[getter]
+    fn limit_early_stopped(&self) -> bool {
+        self.inner.limit_early_stopped
+    }
+    #[getter]
+    fn splits_before_limit(&self) -> usize {
+        self.inner.splits_before_limit
+    }
+    #[getter]
+    fn splits_after_limit(&self) -> usize {
+        self.inner.splits_after_limit
+    }
+    #[getter]
+    fn final_splits(&self) -> usize {
+        self.inner.final_splits
+    }
+    #[getter]
+    fn final_files(&self) -> usize {
+        self.inner.final_files
+    }
+    #[getter]
+    fn planned_data_file_bytes(&self) -> u64 {
+        self.inner.planned_data_file_bytes
+    }
+    #[getter]
+    fn limit(&self) -> Option<usize> {
+        self.inner.limit
+    }
+}
+
 #[pyclass(name = "TableRead", module = "pypaimon_rust.datafusion")]
 pub struct PyTableRead {
     table: Arc<Table>,
@@ -933,16 +1050,12 @@ impl PyRecordBatchReader {
 
 #[pyclass(name = "Plan", module = "pypaimon_rust.datafusion")]
 pub struct PyPlan {
-    splits: Vec<DataSplit>,
-    snapshot_id: Option<i64>,
+    inner: paimon::table::Plan,
 }
 
 impl From<paimon::table::Plan> for PyPlan {
     fn from(plan: paimon::table::Plan) -> Self {
-        Self {
-            splits: plan.splits().to_vec(),
-            snapshot_id: plan.snapshot_id(),
-        }
+        Self { inner: plan }
     }
 }
 
@@ -951,19 +1064,26 @@ impl PyPlan {
     /// Snapshot selected by the scan, even when pruning produces no splits.
     /// `None` means no snapshot was selected, including snapshot-free format tables.
     fn snapshot_id(&self) -> Option<i64> {
-        self.snapshot_id
+        self.inner.snapshot_id()
+    }
+
+    fn stats(&self) -> PyScanStats {
+        PyScanStats {
+            inner: self.inner.stats().clone(),
+        }
     }
 
     fn splits(&self) -> Vec<PySplit> {
-        self.splits
+        self.inner
+            .splits()
             .iter()
             .cloned()
             .map(|inner| PySplit { inner })
             .collect()
     }
 
     fn __len__(&self) -> usize {
-        self.splits.len()
+        self.inner.splits().len()
     }
 }
 

bindings/python/tests/test_read.py

@@ -51,6 +51,39 @@ def _make_variant_table_with_data(warehouse):
     return PaimonCatalog({"warehouse": warehouse}).get_table("vdb.t")
 
 
+@pytest.mark.parametrize("limit,split_count", [(None, 1), (0, 0), (1, 1)])
+def test_scan_statistics_preserve_normal_planning(limit, split_count):
+    with tempfile.TemporaryDirectory() as warehouse:
+        table = _make_table_with_data(warehouse)
+        builder = table.new_read_builder()
+        if limit is not None:
+            builder.with_limit(limit)
+        scan = builder.new_scan()
+        plan = scan.plan()
+        stats = plan.stats()
+        assert len(plan.splits()) == split_count
+        assert plan.snapshot_id() == 1
+        assert stats.manifest_entries_read == (0 if limit == 0 else 1)
+        assert stats.final_splits == stats.final_files == split_count
+        assert stats.limit == limit
+        if limit == 0:
+            assert stats.split_candidates_built == 0
+            assert stats.limit_early_stopped
+        with pytest.raises(AttributeError):
+            stats.manifest_entries_read = 10
+
+
+@pytest.mark.parametrize("start,entry_count", [(0, 1), (1, 0)])
+def test_combined_incremental_scan_statistics(start, entry_count):
+    with tempfile.TemporaryDirectory() as warehouse:
+        table = _make_table_with_data(warehouse)
+        scan = table.new_read_builder().new_incremental_scan(start, 1)
+        plan = scan.plan()
+        stats = plan.stats()
+        assert plan.snapshot_id() == 1
+        assert stats.manifest_entries_read == stats.final_files == entry_count
+
+
 def _variant_read_type(table, columns, paths, fail_on_error=False):
     fields = []
     by_name = {field.name(): field for field in table.schema().fields()}

bindings/python/tests/test_stream_table_scan.py

@@ -62,6 +62,10 @@ def test_cursor_restore_and_initial_vs_follow_up(table, follow_up):
     plan = scan.plan()
     expected_id = 1 if follow_up else 2
     assert plan.snapshot_id() == expected_id
+    stats = plan.stats()
+    assert stats.manifest_entries_read == stats.final_files == (4 if follow_up else 8)
+    assert stats.final_splits == len(plan.splits())
+    assert stats.planned_data_file_bytes > 0
     assert _rows(builder, plan) == [(i, (100 if follow_up else 200) + i) for i in range(32)]
     assert all(split.is_streaming() == follow_up for split in plan.splits())
     assert scan.checkpoint() == expected_id + 1