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 ddc1a55 expose table observability API for pypaimon_rust (#307)
ddc1a55 is described below
commit ddc1a558146c8a38d14bad09b61604e3754c3e2e
Author: SeungMin <[email protected]>
AuthorDate: Mon Jul 13 14:15:23 2026 +0900
expose table observability API for pypaimon_rust (#307)
---
.../python/python/pypaimon_rust/datafusion.pyi | 50 ++++
bindings/python/src/context.rs | 3 +
bindings/python/src/lib.rs | 4 +
bindings/python/src/{lib.rs => partition.rs} | 45 ++--
bindings/python/src/{lib.rs => snapshot.rs} | 47 ++--
bindings/python/src/table.rs | 54 +++++
bindings/python/src/{lib.rs => tag.rs} | 33 ++-
bindings/python/tests/test_datafusion.py | 148 ++++++++++++
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/partition_stat.rs | 269 +++++++++++++++++++++
10 files changed, 614 insertions(+), 41 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index b0d9e4b..07586e6 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -70,12 +70,62 @@ class ReadBuilder:
def new_scan(self) -> TableScan: ...
def new_read(self) -> "TableRead": ...
+# ---- #285: observability ----
+class Snapshot:
+ def id(self) -> int: ...
+ def commit_time_ms(self) -> int: ...
+ def total_record_count(self) -> Optional[int]: ...
+ def delta_record_count(self) -> Optional[int]: ...
+ def commit_kind(self) -> str: ...
+
+class Tag:
+ def name(self) -> str: ...
+ def snapshot_id(self) -> int: ...
+
+class PartitionStat:
+ def partition(self) -> Dict[str, str]: ...
+ def record_count(self) -> int: ...
+ def file_count(self) -> int: ...
+ def total_size_bytes(self) -> int: ...
+
class Table:
def identifier(self) -> str: ...
def location(self) -> str: ...
def schema(self) -> TableSchema: ...
def new_read_builder(self, options: Optional[Dict[str, str]] = None) ->
ReadBuilder: ...
def new_write_builder(self) -> "WriteBuilder": ...
+ def latest_snapshot(self) -> Optional[Snapshot]:
+ """
+ Warning: This method blocks on a DataFusion runtime.
+ Calling this from an active asyncio event loop will result in a panic.
+ """
+ ...
+ def list_snapshots(self) -> List[Snapshot]:
+ """
+ Returns all snapshots ordered newest first (descending by ID).
+
+ Warning: This method blocks on a DataFusion runtime.
+ Calling this from an active asyncio event loop will result in a panic.
+ """
+ ...
+ def list_tags(self) -> List[Tag]:
+ """
+ Warning: This method blocks on a DataFusion runtime.
+ Calling this from an active asyncio event loop will result in a panic.
+ """
+ ...
+ def list_partitions(self) -> List[Dict[str, str]]:
+ """
+ Warning: This method blocks on a DataFusion runtime.
+ Calling this from an active asyncio event loop will result in a panic.
+ """
+ ...
+ def partition_stats(self) -> List[PartitionStat]:
+ """
+ Warning: This method blocks on a DataFusion runtime.
+ Calling this from an active asyncio event loop will result in a panic.
+ """
+ ...
class CommitMessage: ...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index d92d8ee..b2f00f6 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -388,6 +388,9 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<crate::write::PyTableCommit>()?;
this.add_class::<crate::write::PyCommitMessage>()?;
this.add_function(wrap_pyfunction!(udf, &this)?)?;
+ this.add_class::<crate::snapshot::PySnapshot>()?;
+ this.add_class::<crate::tag::PyTag>()?;
+ this.add_class::<crate::partition::PyPartitionStat>()?;
m.add_submodule(&this)?;
py.import("sys")?
.getattr("modules")?
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index dd7177b..f256ff0 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -26,6 +26,10 @@ mod schema;
mod table;
mod udf;
mod write;
+// ---- #285: observability ----
+mod partition;
+mod snapshot;
+mod tag;
#[pymodule]
fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/partition.rs
similarity index 54%
copy from bindings/python/src/lib.rs
copy to bindings/python/src/partition.rs
index dd7177b..f962443 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/partition.rs
@@ -15,20 +15,37 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashMap;
+
+use paimon::table::PartitionStat;
use pyo3::prelude::*;
-mod blob;
-mod context;
-mod error;
-mod predicate;
-mod read;
-mod schema;
-mod table;
-mod udf;
-mod write;
-
-#[pymodule]
-fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
- context::register_module(py, m)?;
- Ok(())
+#[pyclass(name = "PartitionStat", module = "pypaimon_rust.datafusion")]
+pub struct PyPartitionStat {
+ inner: PartitionStat,
+}
+
+impl From<PartitionStat> for PyPartitionStat {
+ fn from(inner: PartitionStat) -> Self {
+ Self { inner }
+ }
+}
+
+#[pymethods]
+impl PyPartitionStat {
+ fn partition(&self) -> HashMap<String, String> {
+ self.inner.partition.clone()
+ }
+
+ fn record_count(&self) -> i64 {
+ self.inner.record_count
+ }
+
+ fn file_count(&self) -> u64 {
+ self.inner.file_count
+ }
+
+ fn total_size_bytes(&self) -> u64 {
+ self.inner.total_size_bytes
+ }
}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/snapshot.rs
similarity index 54%
copy from bindings/python/src/lib.rs
copy to bindings/python/src/snapshot.rs
index dd7177b..5850518 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/snapshot.rs
@@ -15,20 +15,39 @@
// specific language governing permissions and limitations
// under the License.
+use paimon::spec::Snapshot;
use pyo3::prelude::*;
-mod blob;
-mod context;
-mod error;
-mod predicate;
-mod read;
-mod schema;
-mod table;
-mod udf;
-mod write;
-
-#[pymodule]
-fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
- context::register_module(py, m)?;
- Ok(())
+#[pyclass(name = "Snapshot", module = "pypaimon_rust.datafusion")]
+pub struct PySnapshot {
+ inner: Snapshot,
+}
+
+impl PySnapshot {
+ pub fn new(inner: Snapshot) -> Self {
+ Self { inner }
+ }
+}
+
+#[pymethods]
+impl PySnapshot {
+ fn id(&self) -> i64 {
+ self.inner.id()
+ }
+
+ fn commit_time_ms(&self) -> u64 {
+ self.inner.time_millis()
+ }
+
+ fn total_record_count(&self) -> Option<i64> {
+ self.inner.total_record_count()
+ }
+
+ fn delta_record_count(&self) -> Option<i64> {
+ self.inner.delta_record_count()
+ }
+
+ fn commit_kind(&self) -> String {
+ self.inner.commit_kind().to_string()
+ }
}
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index a59ca8f..3fef611 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -15,13 +15,20 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashMap;
use std::sync::Arc;
+use paimon::table::{SnapshotManager, TagManager};
+use paimon_datafusion::runtime::runtime;
use pyo3::prelude::*;
use pyo3::types::PyDict;
+use crate::error::to_py_err;
+use crate::partition::PyPartitionStat;
use crate::read::PyReadBuilder;
use crate::schema::PyTableSchema;
+use crate::snapshot::PySnapshot;
+use crate::tag::PyTag;
use crate::write::PyWriteBuilder;
#[pyclass(name = "Table", module = "pypaimon_rust.datafusion")]
@@ -68,4 +75,51 @@ impl PyTable {
fn new_write_builder(&self) -> PyWriteBuilder {
PyWriteBuilder::new(Arc::clone(&self.inner))
}
+
+ // ---------------- #285: observability ----------------
+ fn latest_snapshot(&self) -> PyResult<Option<PySnapshot>> {
+ let sm = SnapshotManager::new(
+ self.inner.file_io().clone(),
+ self.inner.location().to_string(),
+ );
+ let snap = runtime()
+ .block_on(sm.get_latest_snapshot())
+ .map_err(to_py_err)?;
+ Ok(snap.map(PySnapshot::new))
+ }
+
+ fn list_snapshots(&self) -> PyResult<Vec<PySnapshot>> {
+ let sm = SnapshotManager::new(
+ self.inner.file_io().clone(),
+ self.inner.location().to_string(),
+ );
+ let snaps = runtime().block_on(sm.list_all()).map_err(to_py_err)?;
+ Ok(snaps.into_iter().rev().map(PySnapshot::new).collect())
+ }
+
+ fn list_tags(&self) -> PyResult<Vec<PyTag>> {
+ let tm = TagManager::new(
+ self.inner.file_io().clone(),
+ self.inner.location().to_string(),
+ );
+ let tags = runtime().block_on(tm.list_all()).map_err(to_py_err)?;
+ Ok(tags
+ .into_iter()
+ .map(|(name, snap)| PyTag::new(name, snap.id()))
+ .collect())
+ }
+
+ fn list_partitions(&self) -> PyResult<Vec<HashMap<String, String>>> {
+ let stats = runtime()
+ .block_on(self.inner.partition_stats())
+ .map_err(to_py_err)?;
+ Ok(stats.into_iter().map(|s| s.partition).collect())
+ }
+
+ fn partition_stats(&self) -> PyResult<Vec<PyPartitionStat>> {
+ let stats = runtime()
+ .block_on(self.inner.partition_stats())
+ .map_err(to_py_err)?;
+ Ok(stats.into_iter().map(PyPartitionStat::from).collect())
+ }
}
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/tag.rs
similarity index 67%
copy from bindings/python/src/lib.rs
copy to bindings/python/src/tag.rs
index dd7177b..c679dd0 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/tag.rs
@@ -17,18 +17,25 @@
use pyo3::prelude::*;
-mod blob;
-mod context;
-mod error;
-mod predicate;
-mod read;
-mod schema;
-mod table;
-mod udf;
-mod write;
+#[pyclass(name = "Tag", module = "pypaimon_rust.datafusion")]
+pub struct PyTag {
+ name: String,
+ snapshot_id: i64,
+}
+
+impl PyTag {
+ pub fn new(name: String, snapshot_id: i64) -> Self {
+ Self { name, snapshot_id }
+ }
+}
+
+#[pymethods]
+impl PyTag {
+ fn name(&self) -> String {
+ self.name.clone()
+ }
-#[pymodule]
-fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
- context::register_module(py, m)?;
- Ok(())
+ fn snapshot_id(&self) -> i64 {
+ self.snapshot_id
+ }
}
diff --git a/bindings/python/tests/test_datafusion.py
b/bindings/python/tests/test_datafusion.py
index e68698c..ca4b64a 100644
--- a/bindings/python/tests/test_datafusion.py
+++ b/bindings/python/tests/test_datafusion.py
@@ -868,3 +868,151 @@ def test_list_databases_and_tables():
assert "name" in field_names
# simple_log_table is non-partitioned, so partition keys are empty.
assert schema.partition_keys() == []
+
+# ---------------- #285: observability ----------------
+def test_snapshots_for_simple_table():
+ catalog = PaimonCatalog({"warehouse": WAREHOUSE})
+ table = catalog.get_table("default.simple_log_table")
+
+ snap = table.latest_snapshot()
+ assert snap is not None
+ assert snap.id() >= 1
+ assert snap.commit_time_ms() > 0
+ assert snap.commit_kind() in {"APPEND", "COMPACT", "OVERWRITE", "ANALYZE"}
+
+ snaps = table.list_snapshots()
+ assert len(snaps) >= 1
+ # Newest first.
+ assert snaps[0].id() == snap.id()
+
+
+def test_partitions_and_tags_smoke():
+ catalog = PaimonCatalog({"warehouse": WAREHOUSE})
+ table = catalog.get_table("default.simple_log_table")
+
+ # Non-partitioned, non-tagged table: both should be empty but well-typed.
+ parts = table.list_partitions()
+ stats = table.partition_stats()
+ tags = table.list_tags()
+
+ assert isinstance(parts, list)
+ assert isinstance(stats, list)
+ assert isinstance(tags, list)
+ # simple_log_table has no partition keys -> partition_stats yields a single
+ # empty-partition bucket or zero buckets depending on how the snapshot was
+ # written. Either is acceptable; we just check the shape.
+ for p in parts:
+ assert isinstance(p, dict)
+ for t in tags:
+ assert isinstance(t.name(), str)
+ assert isinstance(t.snapshot_id(), int)
+
+
+def test_partition_stats_with_partitioned_table():
+ """Validates partition_stats() on a real partitioned table created in a
+ temporary warehouse:
+ - list_partitions() returns the correct partition values
+ - partition_stats() returns correct record / file / size counts per
partition
+ This exercises PartitionComputer.generate_part_values and the per-partition
+ aggregation in aggregate_partition_stats.
+ """
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+
+ ctx.sql(
+ "CREATE TABLE paimon.default.events "
+ "(id INT, name STRING, dt STRING) "
+ "PARTITIONED BY (dt)"
+ )
+ ctx.sql(
+ "INSERT INTO paimon.default.events VALUES "
+ "(1, 'alice', '2024-01-01'), "
+ "(2, 'bob', '2024-01-01'), "
+ "(3, 'carol', '2024-01-02')"
+ )
+
+ catalog = PaimonCatalog({"warehouse": warehouse})
+ table = catalog.get_table("default.events")
+
+ # list_partitions() must return exactly the two inserted partitions.
+ parts = table.list_partitions()
+ assert len(parts) == 2
+ part_values = sorted(p["dt"] for p in parts)
+ assert part_values == ["2024-01-01", "2024-01-02"]
+
+ # partition_stats() must report concrete record / file / size counts.
+ stats = table.partition_stats()
+ assert len(stats) == 2
+
+ by_dt = {s.partition()["dt"]: s for s in stats}
+
+ s1 = by_dt["2024-01-01"]
+ assert s1.record_count() == 2
+ assert s1.file_count() >= 1
+ assert s1.total_size_bytes() > 0
+
+ s2 = by_dt["2024-01-02"]
+ assert s2.record_count() == 1
+ assert s2.file_count() >= 1
+ assert s2.total_size_bytes() > 0
+
+ ctx.sql("DROP TABLE paimon.default.events")
+
+
+def test_partition_stats_excludes_overwritten_partition():
+ """Validates the merge_active_entries (FileKind::Add / Delete sign-flip)
logic.
+
+ INSERT OVERWRITE on a specific partition replaces the old data files with
+ new ones. The old files become FileKind::Delete entries in the manifest, so
+ aggregate_partition_stats() must net them out correctly:
+ - The overwritten partition reflects the NEW row count, not the old one.
+ - A partition that is overwritten with zero rows must not appear in the
+ output (file_count nets to 0, triggering the `<= 0` guard).
+ """
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+
+ ctx.sql(
+ "CREATE TABLE paimon.default.events "
+ "(id INT, name STRING, dt STRING) "
+ "PARTITIONED BY (dt)"
+ )
+ # Initial state: 2 rows in 2024-01-01, 1 row in 2024-01-02.
+ ctx.sql(
+ "INSERT INTO paimon.default.events VALUES "
+ "(1, 'alice', '2024-01-01'), "
+ "(2, 'bob', '2024-01-01'), "
+ "(3, 'carol', '2024-01-02')"
+ )
+
+ # Overwrite 2024-01-01 with a single new row.
+ # This generates FileKind::Delete entries for the old files of that
+ # partition and a FileKind::Add entry for the new file.
+ ctx.sql(
+ "INSERT OVERWRITE paimon.default.events "
+ "VALUES (10, 'dave', '2024-01-01')"
+ )
+
+ catalog = PaimonCatalog({"warehouse": warehouse})
+ table = catalog.get_table("default.events")
+
+ # Both partitions must still be present.
+ parts = table.list_partitions()
+ part_values = sorted(p["dt"] for p in parts)
+ assert part_values == ["2024-01-01", "2024-01-02"]
+
+ stats = table.partition_stats()
+ by_dt = {s.partition()["dt"]: s for s in stats}
+
+ # 2024-01-01: overwritten to 1 row — old Delete entries must be
+ # netted out so record_count reflects only the new file.
+ s1 = by_dt["2024-01-01"]
+ assert s1.record_count() == 1
+
+ # 2024-01-02: untouched.
+ s2 = by_dt["2024-01-02"]
+ assert s2.record_count() == 1
+
+ ctx.sql("DROP TABLE paimon.default.events")
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 5a7030a..617c1fb 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -54,6 +54,7 @@ mod kv_file_writer;
mod lumina_index_build_builder;
pub(crate) mod merge_tree_split_generator;
mod partition_filter;
+mod partition_stat;
mod postpone_file_writer;
mod prepared_files;
mod read_builder;
@@ -100,6 +101,7 @@ pub use incremental_scan::{
IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit,
};
pub use lumina_index_build_builder::LuminaIndexBuildBuilder;
+pub use partition_stat::PartitionStat;
pub use read_builder::ReadBuilder;
pub use rest_env::RESTEnv;
pub use scan_trace::ScanTrace;
diff --git a/crates/paimon/src/table/partition_stat.rs
b/crates/paimon/src/table/partition_stat.rs
new file mode 100644
index 0000000..9d78b58
--- /dev/null
+++ b/crates/paimon/src/table/partition_stat.rs
@@ -0,0 +1,269 @@
+// 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.
+
+//! Per-partition statistics computed by scanning the latest snapshot's
manifest entries.
+//!
+//! Mirrors what Java Paimon exposes via the `$partitions` system table for
runtime introspection.
+
+use std::collections::{BTreeMap, HashMap};
+
+use crate::io::FileIO;
+use crate::spec::{
+ avro::from_avro_bytes_fast, merge_active_entries, BinaryRow, CoreOptions,
ManifestEntry,
+ ManifestFileMeta, PartitionComputer, Snapshot,
+};
+use crate::table::SnapshotManager;
+use crate::table::Table;
+
+const MANIFEST_DIR: &str = "manifest";
+
+/// Per-partition aggregated statistics.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct PartitionStat {
+ /// Partition key/value mapping (e.g. `{"dt": "2024-01-01", "hr": "10"}`).
+ pub partition: HashMap<String, String>,
+ /// Total record count across all live data files.
+ pub record_count: i64,
+ /// Live data file count.
+ pub file_count: u64,
+ /// Total bytes for live data files.
+ pub total_size_bytes: u64,
+}
+
+#[derive(Default)]
+struct Accum {
+ record_count: i64,
+ file_count: i64,
+ total_size_bytes: i64,
+}
+
+impl Table {
+ /// Compute per-partition statistics from the latest snapshot.
+ ///
+ /// Mirrors [`catalog::list_partitions_from_file_system`]: calls
+ /// `merge_active_entries` first to deduplicate entries that appear in both
+ /// the base and delta manifests, then aggregates the remaining live ADD
+ /// files in a `BTreeMap` for a deterministic result order.
+ ///
+ /// **Warning:** This method reads all manifest lists and entries from the
latest snapshot.
+ /// For tables with a large number of manifests, this operation can be
expensive.
+ ///
+ /// Returns an empty Vec when the table has no snapshots yet.
+ pub async fn partition_stats(&self) -> crate::Result<Vec<PartitionStat>> {
+ let sm = SnapshotManager::new(self.file_io().clone(),
self.location().to_string());
+ let snapshot = match sm.get_latest_snapshot().await? {
+ Some(s) => s,
+ None => return Ok(Vec::new()),
+ };
+
+ let entries = read_all_manifest_entries(self.file_io(),
self.location(), &snapshot).await?;
+
+ let schema = self.schema();
+ let core = CoreOptions::new(schema.options());
+ let computer = PartitionComputer::new(
+ schema.partition_keys(),
+ schema.fields(),
+ core.partition_default_name(),
+ core.legacy_partition_name(),
+ )?;
+
+ aggregate_partition_stats(entries, &computer)
+ }
+
+ /// List all partition values present in the latest snapshot.
+ ///
+ /// **Warning:** This method computes partition statistics which reads all
manifest lists
+ /// and entries. For large tables, this operation can be expensive.
+ ///
+ /// Returns an empty Vec when the table has no snapshots yet.
+ pub async fn list_partitions(&self) -> crate::Result<Vec<HashMap<String,
String>>> {
+ Ok(self
+ .partition_stats()
+ .await?
+ .into_iter()
+ .map(|s| s.partition)
+ .collect())
+ }
+}
+
+async fn read_manifest_list(
+ file_io: &FileIO,
+ table_path: &str,
+ list_name: &str,
+) -> crate::Result<Vec<ManifestFileMeta>> {
+ if list_name.is_empty() {
+ return Ok(Vec::new());
+ }
+ let path = format!(
+ "{}/{}/{}",
+ table_path.trim_end_matches('/'),
+ MANIFEST_DIR,
+ list_name
+ );
+ let input = file_io.new_input(&path)?;
+ let bytes = input.read().await?;
+ from_avro_bytes_fast::<ManifestFileMeta>(&bytes)
+}
+
+async fn read_all_manifest_entries(
+ file_io: &FileIO,
+ table_path: &str,
+ snapshot: &Snapshot,
+) -> crate::Result<Vec<ManifestEntry>> {
+ let mut metas = read_manifest_list(file_io, table_path,
snapshot.base_manifest_list()).await?;
+ metas.extend(read_manifest_list(file_io, table_path,
snapshot.delta_manifest_list()).await?);
+
+ let manifest_dir = format!("{}/{}", table_path.trim_end_matches('/'),
MANIFEST_DIR);
+ let mut all_entries = Vec::new();
+ for meta in metas {
+ let path = format!("{}/{}", manifest_dir, meta.file_name());
+ let input = file_io.new_input(&path)?;
+ let bytes = input.read().await?;
+ let entries = from_avro_bytes_fast::<ManifestEntry>(&bytes)?;
+ all_entries.extend(entries);
+ }
+ Ok(all_entries)
+}
+
+/// Aggregate live manifest entries into per-partition statistics.
+///
+/// Mirrors `catalog::list_partitions_from_file_system`:
+/// 1. Call `merge_active_entries` to collapse ADD/DELETE pairs and remove
+/// entries shadowed by a later DELETE, including duplicate ADD entries
+/// that may appear in both base and delta manifests after compaction.
+/// 2. Accumulate the remaining live ADD entries in a `BTreeMap` keyed by
+/// raw partition bytes for a deterministic, sorted result order.
+fn aggregate_partition_stats(
+ entries: Vec<ManifestEntry>,
+ computer: &PartitionComputer,
+) -> crate::Result<Vec<PartitionStat>> {
+ // Step 1: deduplicate — collapse ADD/DELETE pairs and drop files that have
+ // been deleted. After this point every remaining entry is a live ADD.
+ let live_entries = merge_active_entries(entries);
+
+ // Step 2: accumulate into a BTreeMap for deterministic partition ordering.
+ let mut grouped: BTreeMap<Vec<u8>, Accum> = BTreeMap::new();
+ for entry in &live_entries {
+ let file = entry.file();
+ let accum = grouped.entry(entry.partition().to_vec()).or_default();
+ accum.record_count += file.row_count;
+ accum.file_count += 1;
+ accum.total_size_bytes += file.file_size;
+ }
+
+ let mut out = Vec::with_capacity(grouped.len());
+ for (partition_bytes, accum) in grouped {
+ let partition = if partition_bytes.is_empty() {
+ HashMap::new()
+ } else {
+ let row = BinaryRow::from_serialized_bytes(&partition_bytes)?;
+ computer.generate_part_values(&row)?.into_iter().collect()
+ };
+ out.push(PartitionStat {
+ partition,
+ record_count: accum.record_count,
+ file_count: accum.file_count as u64,
+ total_size_bytes: accum.total_size_bytes as u64,
+ });
+ }
+ Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{DataFileMeta, FileKind, ManifestEntry};
+
+ /// Build a minimal synthetic ManifestEntry for unit testing.
+ /// Mirrors the helper used in `spec::manifest` tests.
+ fn make_entry(
+ kind: FileKind,
+ partition: Vec<u8>,
+ file_name: &str,
+ row_count: i64,
+ ) -> ManifestEntry {
+ let stats = BinaryTableStats::empty();
+ let file = DataFileMeta {
+ file_name: file_name.to_string(),
+ file_size: row_count * 100,
+ row_count,
+ min_key: vec![],
+ max_key: vec![],
+ key_stats: stats.clone(),
+ value_stats: stats,
+ min_sequence_number: 0,
+ max_sequence_number: 0,
+ schema_id: 0,
+ level: 0,
+ extra_files: vec![],
+ creation_time: None,
+ delete_row_count: None,
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: None,
+ write_cols: None,
+ };
+ ManifestEntry::new(kind, partition, 0, 1, file, 2)
+ }
+
+ /// Duplicate ADD entries (same file appearing in both base and delta
+ /// manifests after compaction) must not be double-counted.
+ /// `merge_active_entries` collapses identical identifiers to one.
+ #[test]
+ fn test_duplicate_add_entries_are_not_double_counted() {
+ let computer =
+ PartitionComputer::new(&[] as &[String], &[], "__DEFAULT_PT__",
false).unwrap();
+
+ // Same file_name / level appears twice (base manifest + delta
manifest).
+ let entries = vec![
+ make_entry(FileKind::Add, vec![], "file-001.parquet", 10),
+ make_entry(FileKind::Add, vec![], "file-001.parquet", 10), //
duplicate
+ ];
+
+ let stats = aggregate_partition_stats(entries, &computer).unwrap();
+
+ // Must produce exactly 1 entry with record_count == 10, not 20.
+ assert_eq!(stats.len(), 1, "expected 1 partition entry");
+ assert_eq!(
+ stats[0].record_count, 10,
+ "duplicate ADD must be collapsed to a single count, not
double-counted"
+ );
+ assert_eq!(stats[0].file_count, 1);
+ }
+
+ /// A DELETE entry that follows an ADD for the same file must cancel it
+ /// out completely; the partition must not appear in the result.
+ #[test]
+ fn test_add_then_delete_removes_partition() {
+ let computer =
+ PartitionComputer::new(&[] as &[String], &[], "__DEFAULT_PT__",
false).unwrap();
+
+ let entries = vec![
+ make_entry(FileKind::Add, vec![], "file-002.parquet", 5),
+ make_entry(FileKind::Delete, vec![], "file-002.parquet", 5),
+ ];
+
+ let stats = aggregate_partition_stats(entries, &computer).unwrap();
+ assert!(
+ stats.is_empty(),
+ "partition should be gone after its only file is deleted"
+ );
+ }
+}
