[Feature] Support full-compaction changelog producer #1320

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Rust can read changelog-producer=full-compaction tables, but it cannot produce their changelog:

• ensure_supported rejects primary-key compaction on these tables (compact.rs#L1222-L1237). As a result, manual compaction and CALL sys.compact fail, and the automatic compaction after each commit is skipped silently (auto_maintenance.rs#L27-L32).
• full-compaction.delta-commits is ignored.

When only Rust-based writers (Rust, Python, DataFusion, C or Go) write to such a table, its level-0 files are never compacted and no changelog is produced. Streaming readers, Java or Rust, start from the max level and then follow the changelog (stream_table_scan.rs#L392-L405), so they see nothing newer than the last Java full compaction.

Solution

Port Java FullChangelogMergeTreeCompactRewriter and FullChangelogMergeFunctionWrapper onto the compaction changelog path from #1252, and add full-compaction triggers.

Changelog semantics

Only compaction into the max level (num-levels - 1) produces changelog. For each key, the merged result is compared with the key's record at the max level:

The changelog-producer.row-deduplicate, changelog-producer.ignore-update-before and changelog-producer.ignore-delete options apply as in Java.

Proposed changes

1. Merge function. Add the full-compaction wrapper, and pass file levels into its merge input.
2. Compaction output. Drop retracts from the max-level output, and accept an empty output when every key was deleted.
3. Planning and commit.
   • Accept the producer in ensure_supported, and treat it like none in planner::enabled.
   • Let validate_inputs and validate_primary_key_changes accept a full selection, which moves all files of a bucket into the max level.
   • Let prepare_results accept max-level changelog.
4. Full-compaction triggers.
   • Manual: a full mode on CompactBuilder, and compact_strategy => 'full' / 'minor' on CALL sys.compact.
   • Automatic: compact_after_write runs a full round when identifier % deltaCommits == 0 || identifier == i64::MAX (Java isFullCompactedIdentifier), where deltaCommits = max(full-compaction.delta-commits, 1). As in Java, every commit runs a full round when the option is unset.

This can land in two PRs: changes 1–3 with the CompactBuilder full mode and docs first, then the remaining triggers.

Anything else?

Out of scope:

• Deletion vectors and first-row: Java rejects both with full-compaction.
• Postpone buckets: Rust compaction does not support them.
• changelog-producer.compaction-interval: a Flink-only option.

Willingness to contribute

• I'm willing to submit a PR!