[JingsongLi]

Purpose

Add a Rust API for consolidating table files and atomically publishing the replacement files in a COMPACT snapshot. Planning, execution and commit are separate so applications can schedule tasks across workers and commit completed subsets.

Related to #286: this provides the Rust compaction prerequisite; Python maintenance bindings remain separate.

Brief change log

• Add CompactBuilder, CompactPlanner, CompactExecutor, serializable plans/tasks/results, and TableCommit::commit_compact.

• Plan unaware append compaction with the batch packing rules from Java AppendCompactCoordinator: select files below target-file-size * compaction.small-file-ratio (default 0.7), sort by size, and accumulate file_size + source.split.open-file-cost to twice the target size. Emit a remaining group only when it meets compaction.min.file-num (default 5). Ordinary read splits retain their separate sequence-ordered max(file_size, open_file_cost) packing.

• Use Java's sequence-window best-effort strategy for unordered bucketed append tables; keep task planning independent of read split boundaries.

• Rewrite complete PK buckets with deduplicate, first-row, partial-update and aggregation, preserving merge state, sequence numbers and tombstones.

• Support fixed and dynamic PK buckets, including cross-partition dynamic buckets. Preserve HASH assignments and carry forward the latest HASH manifest when writes commit concurrently.

• Validate captured inputs, schema and file metadata; reject conflicting rewrites and overlapping positive-level LSM ranges. Bound recovery scans to snapshots after the captured plan, reject plans invalidated by rollback, and keep task-subset retries idempotent. Rebuild file indexes and clean up unsubmitted outputs on execution failure.

• Share a streaming KeyValueDataFileWriter between ordinary PK write-buffer flushes and compaction. Compaction writes ordered physical KV batches directly; sequence allocation and flush-time merging stay in the ordinary writer. Collect key/sequence bounds, statistics and indexes incrementally.

• Separate format-buffer flushing from file rolling. Parquet/ORC/Avro/ROW keep files open across buffer flushes; formats that retain pending rows use bounded rolling as a fallback. Materialize partial output chunks for accurate resource accounting. Avoid orc-rust 0.8 predicate/positional-selection composition for indexed reads across flushed stripes, while preserving exact residual filtering.

Tests

Passed 95 relevant integration tests: 14 append compaction cases, 37 PK compaction cases, and write/resource/statistics/external-path/BLOB/sequence suites. Passed 229 focused unit tests with Vortex enabled, covering Java append planning, typed key ranges, commit/recovery, KV output, statistics and ORC.

cargo test --locked -p paimon --test pk_compact_test --test append_compact_test --test pk_value_stats_test --test writer_resources_test --test table_partial_write_test --test pk_user_sequence_parity_test --test external_data_write_test --test dedicated_blob_write_test --test blob_collection_write_test --test creation_time_write_test
cargo +1.98.0 test --locked -j 4 -p paimon --features vortex --lib -- table::compact::append::tests table::table_commit::tests table::merge_tree_split_generator::tests table::kv_data_file_writer::tests table::kv_file_writer::tests spec::stats::tests arrow::format::orc::tests
cargo +1.98.0 fmt --all -- --check
cargo +1.98.0 clippy --locked -p paimon --lib --test pk_compact_test --test append_compact_test --features vortex -- -D warnings

Regression coverage also compares future writes before/after compaction, rejects non-replayable aggregate retract state, exercises effective aggregator overrides and ignored legacy retracts, checks concurrent positive-level conflicts, and covers rollback and bounded recovery.

Writer regression coverage verifies that tiny write buffers do not fragment Parquet/ORC/Avro/ROW compaction files, indexes and key/sequence bounds span multiple batches, row limits can split one input batch, byte targets roll independently, large input batches do not multiply memory charges, and Mosaic/Vortex retain bounded output buffers. The existing cases cover concurrent writes, snapshot history, merge engines, HASH assignments and migrations, changelog, external paths and failure cleanup.

API and Format

Adds the public Rust compaction API. Task selection and output sizing come from captured table options; there is no per-task file-count cap or size override on the builder. Uses existing data-file, manifest and COMPACT snapshot formats. Planning is stateless for each captured snapshot; streaming coordinator aging and scan paging remain outside this API.

Scope is unaware/unordered bucketed append tables and PK tables with changelog-producer=none/input (first-row requires none). Row tracking, data evolution, deletion vectors, query authorization, rowkind fields, postpone buckets, dedicated blob/vector storage, non-HASH snapshot indexes, and lookup/full-compaction changelog generation remain unsupported. Ordered bucketed append, append incremental clustering, PK clustering override and record-level expiration are explicitly rejected until their Java-specific rewrite contracts are implemented. Reducer groups whose aggregate DELETE/raw/initialization state cannot be replayed are rejected without publication, rather than changing subsequent write results; safe sum/delete cases and ignored partial-update retracts remain supported. PK compaction uses the existing 256 overlapping merge-input limit.

Documentation

Add docs/src/compaction.md and its navigation entry, covering the API workflow, concurrency/retry behavior, merge-engine semantics, dynamic buckets and current limits.

[leaves12138]

LGTM

[jordepic]

Wow what timing I was literally just curious about this