feat(table): add compact fns + binding

Author: nuts-rice

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -61,6 +61,7 @@ class Table:
     def expire_snapshots(self, older_than_ms: int) -> int: ...
     def remove_orphan_files(self) -> int: ...
     def drop_partition(self, partition: Dict[str, Any]) -> None: ...
+    def trigger_compaction(self, full_compact: bool) -> None: ...    
 
 class PaimonCatalog:
     def __init__(self, catalog_options: Dict[str, str]) -> None: ...

bindings/python/src/table.rs

@@ -113,7 +113,9 @@ impl PyTable {
         })
     }
 
-    fn trigger_compaction(&self, _full_compact: bool) -> PyResult<()> {
-        todo!()
+    fn trigger_compaction(&self, py: Python<'_>, full_compact: bool) -> PyResult<()> {
+        let _ = full_compact;
+        let rt = runtime();
+        py.detach(|| rt.block_on(async { self.inner.full_compact().await.map_err(to_py_err) }))
     }
 }

crates/paimon/src/table/mod.rs

@@ -196,6 +196,39 @@ impl Table {
         WriteBuilder::new(self)
     }
 
+    // Perform full compaction by scanning live data. Merge + dedup happen during scan.
+    // Than rewrite it through dynamic OVERWRITE.
+    //
+    // This is whole table operation. When called, it will scan and potentially overwrite all partitions and buckets.
+    // Use with caution.
+    pub async fn full_compaction(&self) -> Result<()> {
+        use futures::StreamExt;
+        let read_builder = self.new_read_builder();
+        let splits = read_builder.new_scan().plan().await?.splits().to_vec();
+        if splits.is_empty() {
+            return Ok(());
+        }
+        let read = read_builder.new_read()?;
+        let mut stream = read.to_arrow(&splits)?;
+        let mut write = self.new_write_builder().with_overwrite().new_write()?;
+        let mut wrote_any = false;
+        while let Some(batch) = stream.next().await {
+            let batch = batch?;
+            if batch.num_rows() == 0 {
+                continue;
+            }
+            write.write_arrow_batch(&batch).await?;
+            wrote_any = true;
+        }
+        if !wrote_any {
+            return Ok(());
+        }
+        let msgs = write.prepare_commit().await?;
+        let commit = self.new_write_builder().new_commit();
+        // None partitions are used for dynamic overwrite.
+        commit.overwrite(msgs, None).await
+    }
+
     /// Create a copy of this table with extra options merged into the schema.
     ///
     /// This never switches the schema version; it corresponds to Java