From 8c94591cb76ec1a6850c5efdae1cc6ce09e884f2 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Tue, 3 Nov 2026 12:56:00 +0800 Subject: [PATCH 1/3] feat(python): expose table compaction through new_compact_builder The compaction APIs merged for #1247 are reachable from DataFusion's CALL sys.compact but not from the Python binding. Expose the same plan-execute-commit flow as a one-shot builder: - with_partition_filter restricts planning to partitions, combining repeated filters with AND, like the read filters. - with_commit_user overrides the generated commit identity. - execute plans all tasks, executes them and submits one commit; a failed task aborts the files already produced so nothing unreferenced is left behind. No commit is made when no task is planned. Tests cover merging with row preservation, the no-task no-commit case, partition isolation and the commit identity, using write-only tables so the explicit compaction has files to merge. Part of #286 --- bindings/python/src/compact.rs | 135 ++++++++++++++++++++++++++ bindings/python/src/context.rs | 1 + bindings/python/src/lib.rs | 1 + bindings/python/src/table.rs | 4 + bindings/python/tests/test_compact.py | 135 ++++++++++++++++++++++++++ docs/src/python-binding.md | 28 ++++++ 6 files changed, 304 insertions(+) create mode 100644 bindings/python/src/compact.rs create mode 100644 bindings/python/tests/test_compact.py diff --git a/bindings/python/src/compact.rs b/bindings/python/src/compact.rs new file mode 100644 index 000000000..b6dd2b6ee --- /dev/null +++ b/bindings/python/src/compact.rs @@ -0,0 +1,135 @@ +// 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 std::sync::Arc; + +use paimon::spec::Predicate; +use paimon::table::{CompactMetrics, Table}; +use paimon_datafusion::runtime::runtime; +use pyo3::prelude::*; +use pyo3::types::PyDict; + +use crate::error::to_py_err; +use crate::predicate::dict_to_table_predicate; + +/// One-shot table compaction, mirroring the `CALL sys.compact` procedure: +/// plan the tasks, execute all of them, then submit everything in a single +/// commit. No commit is made when planning yields no task. +/// +/// The builder only captures configuration; the core `CompactBuilder` is +/// created per execution, so the handle never borrows the table. +#[pyclass(name = "CompactBuilder", module = "pypaimon_rust.datafusion")] +pub struct PyCompactBuilder { + table: Arc, + partition_filters: Vec, + commit_user: Option, +} + +impl PyCompactBuilder { + pub(crate) fn new(table: Arc
) -> Self { + Self { + table, + partition_filters: Vec::new(), + commit_user: None, + } + } +} + +fn metrics_to_map(tasks: usize, metrics: &CompactMetrics) -> HashMap<&'static str, u64> { + HashMap::from([ + ("tasks", tasks as u64), + ("files_removed", metrics.files_removed as u64), + ("files_added", metrics.files_added as u64), + ("rows_written", metrics.rows_written), + ("bytes_written", metrics.bytes_written), + ( + "changelog_files_added", + metrics.changelog_files_added as u64, + ), + ("changelog_rows_written", metrics.changelog_rows_written), + ("changelog_bytes_written", metrics.changelog_bytes_written), + ]) +} + +#[pymethods] +impl PyCompactBuilder { + /// Restrict compaction to the partitions matching the predicate. + /// Repeated calls combine the filters with AND. + fn with_partition_filter<'py>( + mut slf: PyRefMut<'py, Self>, + predicate: &Bound<'_, PyDict>, + ) -> PyResult> { + let predicate = dict_to_table_predicate(predicate, slf.table.schema(), true)?; + slf.partition_filters.push(predicate); + Ok(slf) + } + + /// Set the commit identity of the compaction commit. Defaults to a + /// generated UUID, like the other one-shot maintenance paths. + fn with_commit_user(mut slf: PyRefMut<'_, Self>, user: String) -> PyRefMut<'_, Self> { + slf.commit_user = Some(user); + slf + } + + /// Plan, execute and commit all compaction tasks. + /// + /// Execution failure aborts the files already produced by earlier tasks, + /// so nothing unreferenced is left behind. Returns the task count and the + /// write metrics of the compaction commit. + fn execute(&self, py: Python<'_>) -> PyResult> { + let table = self.table.clone(); + let partition_filters = self.partition_filters.clone(); + let commit_user = self.commit_user.clone(); + py.detach(|| { + runtime().block_on(async move { + let mut builder = table.new_compact_builder(); + for filter in &partition_filters { + builder = builder.with_partition_filter(filter.clone())?; + } + if let Some(user) = &commit_user { + builder = builder.with_commit_user(user.clone())?; + } + + let plan = builder.new_planner()?.plan().await?; + let metrics = if plan.tasks().is_empty() { + CompactMetrics::default() + } else { + let executor = builder.new_executor()?; + let commit = builder.new_commit()?; + let mut results = Vec::with_capacity(plan.tasks().len()); + for task in plan.tasks() { + match executor.execute(task).await { + Ok(result) => results.push(result), + Err(error) => { + // Nothing has been submitted yet, so earlier + // outputs are unreferenced. An abort failure + // must not mask the execution error, like the + // sys.compact procedure. + let _ = commit.abort_compact(&results).await; + return Err(error); + } + } + } + commit.commit_compact(results).await? + }; + Ok(metrics_to_map(plan.tasks().len(), &metrics)) + }) + }) + .map_err(to_py_err) + } +} diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs index 10a952fd6..29bd8607e 100644 --- a/bindings/python/src/context.rs +++ b/bindings/python/src/context.rs @@ -1038,6 +1038,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> this.add_class::()?; this.add_class::()?; this.add_class::()?; + this.add_class::()?; this.add_class::()?; this.add_class::()?; this.add_class::()?; diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs index ddb1b8308..49565faff 100644 --- a/bindings/python/src/lib.rs +++ b/bindings/python/src/lib.rs @@ -22,6 +22,7 @@ mod blob; mod blob_uri_reader; mod catalog_version; mod commit_callback; +mod compact; mod context; mod error; mod evolution; diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs index cf8bafa8b..11b374202 100644 --- a/bindings/python/src/table.rs +++ b/bindings/python/src/table.rs @@ -281,6 +281,10 @@ impl PyTable { PyBatchWriteBuilder::new(Arc::clone(&self.inner)) } + fn new_compact_builder(&self) -> crate::compact::PyCompactBuilder { + crate::compact::PyCompactBuilder::new(Arc::clone(&self.inner)) + } + fn new_stream_scan(&self) -> PyResult { PyReadBuilder::new(Arc::clone(&self.inner)).new_stream_scan() } diff --git a/bindings/python/tests/test_compact.py b/bindings/python/tests/test_compact.py new file mode 100644 index 000000000..e65687cdb --- /dev/null +++ b/bindings/python/tests/test_compact.py @@ -0,0 +1,135 @@ +# 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. +# + +import pyarrow as pa +import pytest + +from pypaimon_rust.datafusion import PaimonCatalog, SQLContext + + +def _register(tmp_path): + context = SQLContext() + context.register_catalog('paimon', {'warehouse': str(tmp_path)}) + return context + + +def _bucket_files(table_dir, partition=None): + base = table_dir / partition if partition else table_dir + return sorted(str(p) for p in base.glob('**/bucket-*/*')) + + +def test_compact_builder_class_is_importable(): + from pypaimon_rust.datafusion import CompactBuilder + + assert CompactBuilder.__name__ == 'CompactBuilder' + + +def test_compact_merges_append_files_and_preserves_rows(tmp_path): + context = _register(tmp_path) + context.sql('CREATE SCHEMA paimon.db') + context.sql("""CREATE TABLE paimon.db.t (id INT, p STRING, v INT) PARTITIONED BY (p) WITH ( + 'target-file-size' = '1mb', 'compaction.min.file-num' = '2', + 'write-only' = 'true')""") + for i in range(3): + context.sql( + f"INSERT INTO paimon.db.t VALUES ({i}, 'a', {i * 10}), ({i + 3}, 'a', {i * 10 + 1})") + + table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') + table_dir = tmp_path / 'db.db' / 't' + assert len(_bucket_files(table_dir)) >= 3 + + metrics = table.new_compact_builder().execute() + + assert metrics['tasks'] >= 1 + assert metrics['files_removed'] >= 2 + assert metrics['files_added'] >= 1 + assert metrics['files_removed'] > metrics['files_added'] + result = pa.Table.from_batches(context.sql('SELECT id, v FROM paimon.db.t ORDER BY id')) + assert result['id'].to_pylist() == [0, 1, 2, 3, 4, 5] + assert result['v'].to_pylist() == [0, 10, 20, 1, 11, 21] + + +def test_compact_without_tasks_makes_no_commit(tmp_path): + context = _register(tmp_path) + context.sql('CREATE SCHEMA paimon.db') + context.sql("""CREATE TABLE paimon.db.t (id INT) WITH ( + 'target-file-size' = '1mb', 'compaction.min.file-num' = '2', + 'write-only' = 'true')""") + context.sql('INSERT INTO paimon.db.t VALUES (1), (2)') + context.sql('INSERT INTO paimon.db.t VALUES (3)') + + table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') + first = table.new_compact_builder().execute() + assert first['tasks'] >= 1 + snapshot = table.latest_snapshot().id() + + second = table.new_compact_builder().execute() + + assert second['tasks'] == 0 + assert second['files_removed'] == 0 + assert second['files_added'] == 0 + assert table.latest_snapshot().id() == snapshot + + +def test_compact_partition_filter_leaves_other_partitions_alone(tmp_path): + context = _register(tmp_path) + context.sql('CREATE SCHEMA paimon.db') + context.sql("""CREATE TABLE paimon.db.t (id INT, p STRING) PARTITIONED BY (p) WITH ( + 'target-file-size' = '1mb', 'compaction.min.file-num' = '2', + 'write-only' = 'true')""") + for i in range(2): + context.sql(f"INSERT INTO paimon.db.t VALUES ({i}, 'a'), ({i}, 'b')") + + table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') + table_dir = tmp_path / 'db.db' / 't' + files_before_b = _bucket_files(table_dir, 'p=b') + + metrics = ( + table.new_compact_builder() + .with_partition_filter({'method': 'equal', 'field': 'p', 'literals': ['a']}) + .execute() + ) + + assert metrics['tasks'] >= 1 + assert metrics['files_removed'] >= 2 + assert _bucket_files(table_dir, 'p=b') == files_before_b + result = pa.Table.from_batches(context.sql('SELECT id, p FROM paimon.db.t ORDER BY p, id')) + assert result['id'].to_pylist() == [0, 1, 0, 1] + assert result['p'].to_pylist() == ['a', 'a', 'b', 'b'] + + +def test_compact_with_commit_user_records_identity(tmp_path): + context = _register(tmp_path) + context.sql('CREATE SCHEMA paimon.db') + context.sql("""CREATE TABLE paimon.db.t (id INT) WITH ( + 'target-file-size' = '1mb', 'compaction.min.file-num' = '2', + 'write-only' = 'true')""") + context.sql('INSERT INTO paimon.db.t VALUES (1), (2)') + context.sql('INSERT INTO paimon.db.t VALUES (3)') + + table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') + metrics = ( + table.new_compact_builder() + .with_commit_user('python-compactor') + .execute() + ) + + assert metrics['tasks'] >= 1 + assert table.latest_snapshot().commit_user().startswith('python-compactor-compact-') + result = pa.Table.from_batches(context.sql('SELECT id FROM paimon.db.t ORDER BY id')) + assert result['id'].to_pylist() == [1, 2, 3] diff --git a/docs/src/python-binding.md b/docs/src/python-binding.md index 0f7eb061c..650401825 100644 --- a/docs/src/python-binding.md +++ b/docs/src/python-binding.md @@ -499,6 +499,34 @@ for stat in table.partition_stats(): print(stat.partition(), stat.record_count(), stat.total_size_bytes()) ``` +## Table Maintenance + +### Compaction + +`table.new_compact_builder()` plans, executes and commits table compaction in +one call, mirroring the `CALL sys.compact` SQL procedure. Nothing is committed +when planning yields no task. A failed task aborts the files already produced +by earlier tasks, so no unreferenced files are left behind. + +```python +metrics = table.new_compact_builder().execute() +print(metrics["tasks"], metrics["files_removed"], metrics["files_added"]) + +# Restrict compaction to partitions matching a predicate (same predicate +# format as read filters), and set the commit identity. +metrics = ( + table.new_compact_builder() + .with_partition_filter({"method": "equal", "field": "dt", "literals": ["2026-11-01"]}) + .with_commit_user("my-compactor") + .execute() +) +``` + +The returned metrics are a dictionary with the task count and the write +statistics of the compaction commit: `tasks`, `files_removed`, `files_added`, +`rows_written`, `bytes_written`, `changelog_files_added`, +`changelog_rows_written` and `changelog_bytes_written`. + ## Python UDF Register Python scalar UDFs into a `SQLContext`: From 2ba4064065ccf80d1492896c5144d72bdbb8f5f5 Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Wed, 4 Nov 2026 14:56:38 +0800 Subject: [PATCH 2/3] fix(python): validate compaction options when set and report abort failures --- .../python/pypaimon_rust/datafusion.pyi | 20 +++ bindings/python/src/compact.rs | 132 ++++++++++++------ bindings/python/tests/test_compact.py | 74 ++++++++++ docs/src/compaction.md | 3 +- docs/src/python-binding.md | 16 ++- 5 files changed, 195 insertions(+), 50 deletions(-) diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 09ed065fb..b8dd3c7c5 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -484,6 +484,7 @@ class Table: def new_global_index_build_builder(self) -> "GlobalIndexBuildBuilder": ... def new_global_index_drop_builder(self) -> "GlobalIndexDropBuilder": ... def new_batch_write_builder(self) -> "BatchWriteBuilder": ... + def new_compact_builder(self) -> "CompactBuilder": ... def new_postpone_fixed_bucket_write_builder(self) -> "PostponeFixedBucketWriteBuilder": ... def new_stream_write_builder(self) -> "StreamWriteBuilder": ... def latest_snapshot(self) -> Optional[Snapshot]: @@ -567,6 +568,25 @@ class GlobalIndexDropBuilder: """Drop matched indexes, or report their count for dry runs.""" ... +class CompactBuilder: + def with_partition_filter(self, predicate: Dict[str, Any]) -> "CompactBuilder": + """Restrict compaction to matching partitions; repeated calls combine with AND.""" + ... + def with_commit_user(self, commit_user: str) -> "CompactBuilder": + """Set the commit identity; snapshots record ``-compact-``.""" + ... + def execute(self) -> Dict[str, int]: + """Plan, execute and commit all tasks; return the task count and write metrics. + + No commit is made when planning yields no task. A failed task's error is + raised after a best-effort deletion of earlier tasks' files; a failed + deletion is reported as a ``RuntimeWarning``. + + Warning: This method blocks on a DataFusion runtime. + Calling this from an active asyncio event loop will result in a panic. + """ + ... + class CommitMessage: @staticmethod def deserialize(data: bytes, *, version: int = 14) -> "CommitMessage": diff --git a/bindings/python/src/compact.rs b/bindings/python/src/compact.rs index b6dd2b6ee..99ac4436b 100644 --- a/bindings/python/src/compact.rs +++ b/bindings/python/src/compact.rs @@ -21,6 +21,7 @@ use std::sync::Arc; use paimon::spec::Predicate; use paimon::table::{CompactMetrics, Table}; use paimon_datafusion::runtime::runtime; +use pyo3::exceptions::PyRuntimeWarning; use pyo3::prelude::*; use pyo3::types::PyDict; @@ -32,7 +33,7 @@ use crate::predicate::dict_to_table_predicate; /// commit. No commit is made when planning yields no task. /// /// The builder only captures configuration; the core `CompactBuilder` is -/// created per execution, so the handle never borrows the table. +/// created per call, so the handle never borrows the table. #[pyclass(name = "CompactBuilder", module = "pypaimon_rust.datafusion")] pub struct PyCompactBuilder { table: Arc
, @@ -75,61 +76,102 @@ impl PyCompactBuilder { predicate: &Bound<'_, PyDict>, ) -> PyResult> { let predicate = dict_to_table_predicate(predicate, slf.table.schema(), true)?; + // Validation belongs to the core builder; each filter is checked alone. + slf.table + .new_compact_builder() + .with_partition_filter(predicate.clone()) + .map_err(to_py_err)?; slf.partition_filters.push(predicate); Ok(slf) } /// Set the commit identity of the compaction commit. Defaults to a - /// generated UUID, like the other one-shot maintenance paths. - fn with_commit_user(mut slf: PyRefMut<'_, Self>, user: String) -> PyRefMut<'_, Self> { - slf.commit_user = Some(user); - slf + /// generated UUID. The snapshot records it as `-compact-`. + fn with_commit_user( + mut slf: PyRefMut<'_, Self>, + commit_user: String, + ) -> PyResult> { + slf.table + .new_compact_builder() + .with_commit_user(commit_user.clone()) + .map_err(to_py_err)?; + slf.commit_user = Some(commit_user); + Ok(slf) } /// Plan, execute and commit all compaction tasks. /// - /// Execution failure aborts the files already produced by earlier tasks, - /// so nothing unreferenced is left behind. Returns the task count and the - /// write metrics of the compaction commit. + /// When a task fails, the files written by earlier tasks are deleted on a + /// best-effort basis; a failed deletion is reported as a `RuntimeWarning` + /// and the task's error is raised. Returns the task count and the write + /// metrics of the compaction commit. fn execute(&self, py: Python<'_>) -> PyResult> { - let table = self.table.clone(); - let partition_filters = self.partition_filters.clone(); - let commit_user = self.commit_user.clone(); - py.detach(|| { - runtime().block_on(async move { - let mut builder = table.new_compact_builder(); - for filter in &partition_filters { - builder = builder.with_partition_filter(filter.clone())?; - } - if let Some(user) = &commit_user { - builder = builder.with_commit_user(user.clone())?; + let mut abort_error = None; + let result = py.detach(|| { + runtime().block_on(compact( + &self.table, + &self.partition_filters, + self.commit_user.as_deref(), + &mut abort_error, + )) + }); + if let Some(error) = abort_error { + warn_abort_failure(py, &self.table, &error); + } + result.map_err(to_py_err) + } +} + +async fn compact( + table: &Table, + partition_filters: &[Predicate], + commit_user: Option<&str>, + abort_error: &mut Option, +) -> paimon::Result> { + let mut builder = table.new_compact_builder(); + for filter in partition_filters { + builder = builder.with_partition_filter(filter.clone())?; + } + if let Some(user) = commit_user { + builder = builder.with_commit_user(user)?; + } + + let plan = builder.new_planner()?.plan().await?; + let metrics = if plan.tasks().is_empty() { + CompactMetrics::default() + } else { + let executor = builder.new_executor()?; + let commit = builder.new_commit()?; + let mut results = Vec::with_capacity(plan.tasks().len()); + for task in plan.tasks() { + match executor.execute(task).await { + Ok(result) => results.push(result), + Err(error) => { + // Nothing has been submitted yet, so earlier outputs are + // unreferenced. Keep the task error as the one raised. + *abort_error = commit.abort_compact(&results).await.err(); + return Err(error); } + } + } + commit.commit_compact(results).await? + }; + Ok(metrics_to_map(plan.tasks().len(), &metrics)) +} - let plan = builder.new_planner()?.plan().await?; - let metrics = if plan.tasks().is_empty() { - CompactMetrics::default() - } else { - let executor = builder.new_executor()?; - let commit = builder.new_commit()?; - let mut results = Vec::with_capacity(plan.tasks().len()); - for task in plan.tasks() { - match executor.execute(task).await { - Ok(result) => results.push(result), - Err(error) => { - // Nothing has been submitted yet, so earlier - // outputs are unreferenced. An abort failure - // must not mask the execution error, like the - // sys.compact procedure. - let _ = commit.abort_compact(&results).await; - return Err(error); - } - } - } - commit.commit_compact(results).await? - }; - Ok(metrics_to_map(plan.tasks().len(), &metrics)) - }) - }) - .map_err(to_py_err) +/// The task error is still raised, so a failing `warnings.warn` (for example +/// under `-W error`) is ignored rather than replacing it. +fn warn_abort_failure(py: Python<'_>, table: &Table, error: &paimon::Error) { + if let Ok(warnings) = py.import("warnings") { + let _ = warnings.call_method1( + "warn", + ( + format!( + "Failed to delete unsubmitted compaction files of table {}: {error}", + table.identifier().full_name() + ), + py.get_type::(), + ), + ); } } diff --git a/bindings/python/tests/test_compact.py b/bindings/python/tests/test_compact.py index e65687cdb..c1501e3fb 100644 --- a/bindings/python/tests/test_compact.py +++ b/bindings/python/tests/test_compact.py @@ -33,6 +33,30 @@ def _bucket_files(table_dir, partition=None): return sorted(str(p) for p in base.glob('**/bucket-*/*')) +def _assert_compact_snapshot(table, previous): + snapshot = table.latest_snapshot() + assert snapshot.id() == previous + 1 + assert snapshot.commit_kind() == 'COMPACT' + + +def _write_primary_key_table(tmp_path, options=None): + context = _register(tmp_path) + context.sql('CREATE SCHEMA paimon.db') + options = {'bucket': '1', 'write-only': 'true', **(options or {})} + properties = ', '.join(f"'{name}' = '{value}'" for name, value in options.items()) + context.sql( + f'CREATE TABLE paimon.db.t (id INT, v INT, PRIMARY KEY (id)) WITH ({properties})') + context.sql('INSERT INTO paimon.db.t VALUES (1, 10), (2, 20)') + context.sql('INSERT INTO paimon.db.t VALUES (1, 11)') + + +def _read_rows(table): + read = table.new_read_builder() + return sorted((row['id'], row['v']) + for batch in read.new_read().read(read.new_scan().plan().splits()) + for row in batch.to_pylist()) + + def test_compact_builder_class_is_importable(): from pypaimon_rust.datafusion import CompactBuilder @@ -52,6 +76,7 @@ def test_compact_merges_append_files_and_preserves_rows(tmp_path): table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') table_dir = tmp_path / 'db.db' / 't' assert len(_bucket_files(table_dir)) >= 3 + snapshot = table.latest_snapshot().id() metrics = table.new_compact_builder().execute() @@ -59,6 +84,7 @@ def test_compact_merges_append_files_and_preserves_rows(tmp_path): assert metrics['files_removed'] >= 2 assert metrics['files_added'] >= 1 assert metrics['files_removed'] > metrics['files_added'] + _assert_compact_snapshot(table, snapshot) result = pa.Table.from_batches(context.sql('SELECT id, v FROM paimon.db.t ORDER BY id')) assert result['id'].to_pylist() == [0, 1, 2, 3, 4, 5] assert result['v'].to_pylist() == [0, 10, 20, 1, 11, 21] @@ -97,7 +123,9 @@ def test_compact_partition_filter_leaves_other_partitions_alone(tmp_path): table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') table_dir = tmp_path / 'db.db' / 't' + files_before_a = _bucket_files(table_dir, 'p=a') files_before_b = _bucket_files(table_dir, 'p=b') + snapshot = table.latest_snapshot().id() metrics = ( table.new_compact_builder() @@ -107,6 +135,8 @@ def test_compact_partition_filter_leaves_other_partitions_alone(tmp_path): assert metrics['tasks'] >= 1 assert metrics['files_removed'] >= 2 + _assert_compact_snapshot(table, snapshot) + assert set(files_before_a) < set(_bucket_files(table_dir, 'p=a')) assert _bucket_files(table_dir, 'p=b') == files_before_b result = pa.Table.from_batches(context.sql('SELECT id, p FROM paimon.db.t ORDER BY p, id')) assert result['id'].to_pylist() == [0, 1, 0, 1] @@ -133,3 +163,47 @@ def test_compact_with_commit_user_records_identity(tmp_path): assert table.latest_snapshot().commit_user().startswith('python-compactor-compact-') result = pa.Table.from_batches(context.sql('SELECT id FROM paimon.db.t ORDER BY id')) assert result['id'].to_pylist() == [1, 2, 3] + + +def test_compact_builder_validates_arguments_when_set(tmp_path): + context = _register(tmp_path) + context.sql('CREATE SCHEMA paimon.db') + context.sql('CREATE TABLE paimon.db.t (id INT, p STRING) PARTITIONED BY (p)') + builder = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t').new_compact_builder() + + with pytest.raises(ValueError, match='partition keys'): + builder.with_partition_filter({'method': 'equal', 'field': 'id', 'literals': [1]}) + with pytest.raises(ValueError, match='commit_user'): + builder.with_commit_user(commit_user='a/b') + + +def test_compact_merges_primary_key_sorted_runs(tmp_path): + _write_primary_key_table(tmp_path) + table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') + snapshot = table.latest_snapshot().id() + + metrics = table.new_compact_builder().execute() + + assert metrics['tasks'] == 1 + assert metrics['files_removed'] == 2 + assert metrics['files_added'] == 1 + _assert_compact_snapshot(table, snapshot) + assert _read_rows(table) == [(1, 11), (2, 20)] + + +def test_compact_deletion_vector_table_requires_local_cache(tmp_path): + _write_primary_key_table(tmp_path, {'deletion-vectors.enabled': 'true'}) + table = PaimonCatalog({'warehouse': str(tmp_path)}).get_table('db.t') + snapshot = table.latest_snapshot().id() + + with pytest.raises(ValueError, match='local-cache.enabled'): + table.new_compact_builder().execute() + assert table.latest_snapshot().id() == snapshot + + cached = PaimonCatalog({'warehouse': str(tmp_path), 'local-cache.enabled': 'true'}) + table = cached.get_table('db.t') + # Write-only deletion-vector tables expose level-0 rows only after compaction. + assert _read_rows(table) == [] + table.new_compact_builder().execute() + _assert_compact_snapshot(table, snapshot) + assert _read_rows(table) == [(1, 11), (2, 20)] diff --git a/docs/src/compaction.md b/docs/src/compaction.md index c1c5eaff8..1b24e652f 100644 --- a/docs/src/compaction.md +++ b/docs/src/compaction.md @@ -25,7 +25,8 @@ tables with fixed or dynamic buckets. `CompactBuilder` creates a `CompactPlanner`, a `CompactExecutor`, and a `TableCommit`. Planning and execution do not publish snapshots; `commit_compact` atomically publishes the selected results in a `COMPACT` snapshot. DataFusion SQL runs this flow through -[`CALL sys.compact`](sql.md#compact). +[`CALL sys.compact`](sql.md#compact), and Python through +[`table.new_compact_builder()`](python-binding.md#compaction). The builder selects a dedicated append, primary-key or Data Evolution planner for the table mode. They share snapshot capture and the public plan/task types; diff --git a/docs/src/python-binding.md b/docs/src/python-binding.md index 2190a98a8..57b8ac555 100644 --- a/docs/src/python-binding.md +++ b/docs/src/python-binding.md @@ -524,16 +524,18 @@ for stat in table.partition_stats(): ### Compaction `table.new_compact_builder()` plans, executes and commits table compaction in -one call, mirroring the `CALL sys.compact` SQL procedure. Nothing is committed -when planning yields no task. A failed task aborts the files already produced -by earlier tasks, so no unreferenced files are left behind. +one call, like the [`CALL sys.compact`](sql.md#compact) SQL procedure. Nothing +is committed when planning yields no task. When a task fails, the files written +by earlier tasks are deleted on a best-effort basis; a failed deletion is +reported as a `RuntimeWarning` before the task's error is raised. ```python metrics = table.new_compact_builder().execute() print(metrics["tasks"], metrics["files_removed"], metrics["files_added"]) # Restrict compaction to partitions matching a predicate (same predicate -# format as read filters), and set the commit identity. +# format as read filters), and set the commit user. It defaults to a generated +# UUID; the snapshot records it as `-compact-`. metrics = ( table.new_compact_builder() .with_partition_filter({"method": "equal", "field": "dt", "literals": ["2026-11-01"]}) @@ -547,6 +549,12 @@ statistics of the compaction commit: `tasks`, `files_removed`, `files_added`, `rows_written`, `bytes_written`, `changelog_files_added`, `changelog_rows_written` and `changelog_bytes_written`. +Compacting a primary-key table that uses the `lookup` changelog producer, the +`first-row` merge engine or deletion vectors needs `local-cache.enabled=true` in +the catalog options; without it `execute()` raises `ValueError` once there are +tasks to run. See [Table Compaction](compaction.md) for the supported table +modes. + ## Python UDF Register Python scalar UDFs into a `SQLContext`: From eb3fb04fdc37270ecd93c228dc6ea2bf40bc136b Mon Sep 17 00:00:00 2001 From: yangjie01 Date: Thu, 5 Nov 2026 18:03:05 +0800 Subject: [PATCH 3/3] docs(python): describe execute() blocking in asyncio Co-Authored-By: Claude Code --- bindings/python/python/pypaimon_rust/datafusion.pyi | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index b8dd3c7c5..19a8af3dc 100644 --- a/bindings/python/python/pypaimon_rust/datafusion.pyi +++ b/bindings/python/python/pypaimon_rust/datafusion.pyi @@ -582,8 +582,9 @@ class CompactBuilder: raised after a best-effort deletion of earlier tasks' files; a failed deletion is reported as a ``RuntimeWarning``. - Warning: This method blocks on a DataFusion runtime. - Calling this from an active asyncio event loop will result in a panic. + Warning: This method blocks the calling thread until it returns, with the + GIL released. Called from an asyncio event loop, it blocks the loop; use + ``await asyncio.to_thread(builder.execute)`` to keep the loop running. """ ...