diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi index 09ed065fb..19a8af3dc 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,26 @@ 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 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. + """ + ... + 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 new file mode 100644 index 000000000..99ac4436b --- /dev/null +++ b/bindings/python/src/compact.rs @@ -0,0 +1,177 @@ +// 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::exceptions::PyRuntimeWarning; +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 call, 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)?; + // 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. 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. + /// + /// 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 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)) +} + +/// 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/src/context.rs b/bindings/python/src/context.rs index aa11040c6..75e16b372 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..c1501e3fb --- /dev/null +++ b/bindings/python/tests/test_compact.py @@ -0,0 +1,209 @@ +# 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 _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 + + 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 + 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 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] + + +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_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() + .with_partition_filter({'method': 'equal', 'field': 'p', 'literals': ['a']}) + .execute() + ) + + 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] + 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] + + +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 8ac271136..57b8ac555 100644 --- a/docs/src/python-binding.md +++ b/docs/src/python-binding.md @@ -519,6 +519,42 @@ 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, 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 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"]}) + .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`. + +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`: