expose table observability API for pypaimon_rust

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -59,12 +59,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: ...
 

bindings/python/src/context.rs

@@ -382,6 +382,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")?

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<()> {

bindings/python/src/partition.rs

@@ -0,0 +1,51 @@
+// 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.
+
+use std::collections::HashMap;
+
+use paimon::table::PartitionStat;
+use pyo3::prelude::*;
+
+#[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
+    }
+}

bindings/python/src/snapshot.rs

@@ -0,0 +1,53 @@
+// 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.
+
+use paimon::spec::Snapshot;
+use pyo3::prelude::*;
+
+#[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()
+    }
+}

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())
+    }
 }

bindings/python/src/tag.rs

@@ -0,0 +1,41 @@
+// 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.
+
+use pyo3::prelude::*;
+
+#[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()
+    }
+
+    fn snapshot_id(&self) -> i64 {
+        self.snapshot_id
+    }
+}