From 590f8292ea43525ea0918ddfb304832397eecdc0 Mon Sep 17 00:00:00 2001 From: nuts_rice Date: Tue, 30 Jun 2026 09:48:03 -0600 Subject: [PATCH 1/4] feat(table): add table managment fns WIP --- bindings/python/src/table.rs | 21 +++++++++-- crates/paimon/src/table/snapshot_manager.rs | 40 +++++++++++++++++++++ 2 files changed, 58 insertions(+), 3 deletions(-) diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs index 263c821c..7b20bbfe 100644 --- a/bindings/python/src/table.rs +++ b/bindings/python/src/table.rs @@ -14,10 +14,9 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. - -use std::sync::Arc; - use pyo3::prelude::*; +use std::collections::HashMap; +use std::sync::Arc; use crate::read::PyReadBuilder; use crate::schema::PyTableSchema; @@ -52,4 +51,20 @@ impl PyTable { fn new_read_builder(&self) -> PyReadBuilder { PyReadBuilder::new(Arc::clone(&self.inner)) } + + fn expire_snapshots(&self, cutoff_time: i64) -> PyResult { + todo!() + } + + fn remove_orphan_files(&self, cutoff_time: i64) -> PyResult { + todo!() + } + + fn drop_partition(&self, partition: HashMap) -> PyResult<()> { + todo!() + } + + fn trigger_compaction(&self, full_compact: bool) -> PyResult<()> { + todo!() + } } diff --git a/crates/paimon/src/table/snapshot_manager.rs b/crates/paimon/src/table/snapshot_manager.rs index 73effe69..30801c09 100644 --- a/crates/paimon/src/table/snapshot_manager.rs +++ b/crates/paimon/src/table/snapshot_manager.rs @@ -345,6 +345,46 @@ impl SnapshotManager { Ok(result) } + pub async fn expire_snapshots(&self, snapshot_ids: &[i64]) -> crate::Result { + let mut deleted_count = 0; + for &snapshot_id in snapshot_ids { + self.delete_snapshot(snapshot_id).await?; + deleted_count += 1; + } + Ok(deleted_count) + } + + // Expires snapshots whose commit time is earlier than target timestamp_millis. + // Returns the number of snapshots deleted. + pub async fn expire_snapshots_earlier_than(&self, timestamp_millis: i64) -> crate::Result { + let snapshot_ids = self.list_all_ids().await?; + let mut to_delete = Vec::new(); + for snapshot_id in snapshot_ids { + let snapshot = self.get_snapshot(snapshot_id).await?; + if (snapshot.time_millis() as i64) < timestamp_millis { + to_delete.push(snapshot_id); + } else { + break; + } + } + self.expire_snapshots(&to_delete).await?; + Ok(to_delete.len() as i64) + } + + // Remove files which are not used by any snapshots and are not referenced by + // manifest lists. + pub async fn remove_orphan_files(&self) -> crate::Result<()> { + let snapshot_ids = self.list_all_ids().await?; + let mut manifest_files = std::collections::HashSet::new(); + for snap_id in snapshot_ids { + let snapshot = self.get_snapshot(snap_id).await?; + manifest_files.insert(snapshot.base_manifest_list().to_string()); + manifest_files.insert(snapshot.delta_manifest_list().to_string()); + } + + Ok(()) + } + /// Returns the snapshot whose commit time is earlier than or equal to the given /// `timestamp_millis`. If no such snapshot exists, returns None. /// From 9f9c96cc6b56688c91c838ca8f16bb863360cac5 Mon Sep 17 00:00:00 2001 From: nuts_rice Date: Tue, 30 Jun 2026 15:38:24 -0600 Subject: [PATCH 2/4] feat(table): work on table managment fn bindings --- .../python/pypaimon_rust/datafusion.pyi | 3 + bindings/python/src/table.rs | 58 ++++++++++++++++--- crates/paimon/src/table/snapshot_manager.rs | 14 +++++ 3 files changed, 67 insertions(+), 8 deletions(-) diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 772f31fe..7206f7ff 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -58,6 +58,9 @@ class Table: def location(self) -> str: ... def schema(self) -> TableSchema: ... def new_read_builder(self) -> ReadBuilder: ... + def expire_snapshots(self, older_than_ms: int) -> int: ... + def remove_orphan_files(self) -> int: ... + def drop_partition(self, partition: Dict[str, str]) -> None: ... class PaimonCatalog: def __init__(self, catalog_options: Dict[str, str]) -> None: ... diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs index 7b20bbfe..3fb59f05 100644 --- a/bindings/python/src/table.rs +++ b/bindings/python/src/table.rs @@ -16,6 +16,10 @@ // under the License. use pyo3::prelude::*; use std::collections::HashMap; + +use paimon::spec::Datum; +use paimon::table::SnapShotManager; +use paimon_datafusion::runtime::runtime; use std::sync::Arc; use crate::read::PyReadBuilder; @@ -52,18 +56,56 @@ impl PyTable { PyReadBuilder::new(Arc::clone(&self.inner)) } - fn expire_snapshots(&self, cutoff_time: i64) -> PyResult { - todo!() + fn expire_snapshots(&self, py: Python<'_>, older_than_ms: i64) -> PyResult { + let rt = runtime(); + py.detach(|| { + rt.block_on(async { + let snapshot_manager = SnapShotManager::new(Arc::clone(&self.inner)); + snapshot_manager + .expire_snapshots_earlier_than(older_than_millis) + .await + .map_err(to_py_err) + }) + }) } - - fn remove_orphan_files(&self, cutoff_time: i64) -> PyResult { - todo!() + fn remove_orphan_files(&self, py: Python<'_>) -> PyResult { + let rt = runtime(); + py.detach(|| { + rt.block_on(async { + let snapshot_manager = SnapshotManager::new( + self.inner.file_io().clone(), + self.inner.location().to_string(), + ); + snapshot_manager + .remove_orphan_files() + .await + .map_err(to_py_err) + }) + }) } - fn drop_partition(&self, partition: HashMap) -> PyResult<()> { - todo!() + fn drop_partition(&self, partition: HashMap>) -> PyResult<()> { + let partition_fields = self.inner.schema().partition_fields(); + let mut spec: HashMap> = HashMap::with_capacity(partition.len()); + for (k, v) in &partition { + let datum = if v.is_none() { + None + } else { + let field = partition_fields + .iter() + .find(|f| f.name() == k) + .ok_or_else(|| { + PyValueError::new_err(format!("Partition field {} not found in schema", k)) + })?; + Some(py_to_datum(v, field.data_type())?) + }; + spec.insert(k.clone(), datum); + } + runtime().block_on(async { + let commit = self.inner.new_write_builder().new_commit(); + commit.drop_partitions(vec![spec]).await.map_err(to_py_err) + }) } - fn trigger_compaction(&self, full_compact: bool) -> PyResult<()> { todo!() } diff --git a/crates/paimon/src/table/snapshot_manager.rs b/crates/paimon/src/table/snapshot_manager.rs index 30801c09..6702ac05 100644 --- a/crates/paimon/src/table/snapshot_manager.rs +++ b/crates/paimon/src/table/snapshot_manager.rs @@ -381,6 +381,20 @@ impl SnapshotManager { manifest_files.insert(snapshot.base_manifest_list().to_string()); manifest_files.insert(snapshot.delta_manifest_list().to_string()); } + let manifest_dir = self.manifest_dir(); + let statuses = self.file_io.list_status(&manifest_dir).await?; + let mut deleted_count = 0; + for status in statuses { + if status.is_dir() { + continue; + } + let name = status.path.rsplit('/').next().unwrap_or(&status.path); + let manifest_path = format!("{}/{}", manifest_dir, name); + if !manifest_files.contains(&manifest_path) { + self.file_io.delete_file(&manifest_path).await?; + deleted_count += 1; + } + } Ok(()) } From 956b185a591557d4a2427854dd502832702217b2 Mon Sep 17 00:00:00 2001 From: nuts_rice Date: Thu, 2 Jul 2026 10:36:24 -0600 Subject: [PATCH 3/4] fix: fix bindings and rm_orphan_files --- bindings/python/src/table.rs | 19 +++++++++++++------ crates/paimon/src/table/snapshot_manager.rs | 6 +++--- 2 files changed, 16 insertions(+), 9 deletions(-) diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs index 3fb59f05..37a774fa 100644 --- a/bindings/python/src/table.rs +++ b/bindings/python/src/table.rs @@ -14,14 +14,17 @@ // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. +use pyo3::exceptions::PyValueError; use pyo3::prelude::*; use std::collections::HashMap; +use std::sync::Arc; use paimon::spec::Datum; -use paimon::table::SnapShotManager; +use paimon::table::SnapshotManager; use paimon_datafusion::runtime::runtime; -use std::sync::Arc; +use crate::error::to_py_err; +use crate::predicate::py_to_datum; use crate::read::PyReadBuilder; use crate::schema::PyTableSchema; @@ -55,19 +58,22 @@ impl PyTable { fn new_read_builder(&self) -> PyReadBuilder { PyReadBuilder::new(Arc::clone(&self.inner)) } - fn expire_snapshots(&self, py: Python<'_>, older_than_ms: i64) -> PyResult { let rt = runtime(); py.detach(|| { rt.block_on(async { - let snapshot_manager = SnapShotManager::new(Arc::clone(&self.inner)); + let snapshot_manager = SnapshotManager::new( + self.inner.file_io().clone(), + self.inner.location().to_string(), + ); snapshot_manager - .expire_snapshots_earlier_than(older_than_millis) + .expire_snapshots_earlier_than(older_than_ms) .await .map_err(to_py_err) }) }) } + fn remove_orphan_files(&self, py: Python<'_>) -> PyResult { let rt = runtime(); py.detach(|| { @@ -106,7 +112,8 @@ impl PyTable { commit.drop_partitions(vec![spec]).await.map_err(to_py_err) }) } - fn trigger_compaction(&self, full_compact: bool) -> PyResult<()> { + + fn trigger_compaction(&self, _full_compact: bool) -> PyResult<()> { todo!() } } diff --git a/crates/paimon/src/table/snapshot_manager.rs b/crates/paimon/src/table/snapshot_manager.rs index 6702ac05..1b5706c1 100644 --- a/crates/paimon/src/table/snapshot_manager.rs +++ b/crates/paimon/src/table/snapshot_manager.rs @@ -373,7 +373,7 @@ impl SnapshotManager { // Remove files which are not used by any snapshots and are not referenced by // manifest lists. - pub async fn remove_orphan_files(&self) -> crate::Result<()> { + pub async fn remove_orphan_files(&self) -> crate::Result { let snapshot_ids = self.list_all_ids().await?; let mut manifest_files = std::collections::HashSet::new(); for snap_id in snapshot_ids { @@ -385,7 +385,7 @@ impl SnapshotManager { let statuses = self.file_io.list_status(&manifest_dir).await?; let mut deleted_count = 0; for status in statuses { - if status.is_dir() { + if status.is_dir { continue; } let name = status.path.rsplit('/').next().unwrap_or(&status.path); @@ -396,7 +396,7 @@ impl SnapshotManager { } } - Ok(()) + Ok(deleted_count) } /// Returns the snapshot whose commit time is earlier than or equal to the given From 0dca5cebeab12a8148172c7c5d9c1eb0ec14369c Mon Sep 17 00:00:00 2001 From: nuts_rice Date: Mon, 6 Jul 2026 08:59:14 -0600 Subject: [PATCH 4/4] fix: fix drop_partition py binding --- bindings/python/python/pypaimon_rust/datafusion.pyi | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 7206f7ff..c023d5d3 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -60,7 +60,7 @@ class Table: def new_read_builder(self) -> ReadBuilder: ... def expire_snapshots(self, older_than_ms: int) -> int: ... def remove_orphan_files(self) -> int: ... - def drop_partition(self, partition: Dict[str, str]) -> None: ... + def drop_partition(self, partition: Dict[str, Any]) -> None: ... class PaimonCatalog: def __init__(self, catalog_options: Dict[str, str]) -> None: ...