[Core] Add compaction.task-threads for multi-thread async compaction per write subtask

[jacklong319]

Purpose

Linked issue: close #10132

Today each Flink write subtask uses one shared async compaction thread for all (partition, bucket) writers. Under peak load, compaction falls behind; in Deletion Vectors (MOW) mode, delayed L0 compaction hurts read freshness (see #10132).

This PR adds table option compaction.task-threads to run parallel async compaction across buckets within the same write subtask, while compaction for the same bucket remains serialized.

Semantics:

• 1 (default): SINGLE — unchanged behavior.
• N > 1: FIXED_POOL — N threads shared by buckets in the subtask.
• -1: PER_BUCKET — one thread per active (partition, bucket) writer (higher memory use).

Changes:

• Add CompactionTaskExecutorMode and CoreOptions.COMPACTION_TASK_THREADS.
• Route executors in AbstractFileStoreWrite.compactExecutor(partition, bucket); release per-bucket executors on writer cleanup where applicable.
• Default remains 1; no storage format change.

Tests

• mvn -pl paimon-api,paimon-core -am -Pfast-build test
• Unit tests for executor mode selection (if included in this PR)
• Existing write/compaction tests pass
• Manual validation as described in [Feature] Configurable multi-thread async compaction per Flink write subtask (compaction.task-threads) #10132 (optional; production summary stays in the issue)

[JingsongLi]

The linked production issue provides a strong end-to-end case for cross-bucket compaction parallelism (reported DV read lag improved from ~40 min at 1 thread to ~8 min at 3). I traced the executor routing and the per-bucket compact managers’ taskFuture guard; the default remains a single shared executor, and the managers appear to keep each bucket’s submitted task serialized.

Merge blockers / production follow-up:

1. A normal JDK 8 mvn -o -pl paimon-api,paimon-core -DskipTests compile fails Spotless in AbstractFileStoreWrite.java at the perBucketCompactExecutors.remove(...) line. This must be formatted; the PR currently has 14 failing checks.
2. There are no new tests for FIXED_POOL or PER_BUCKET, cross-bucket concurrency, same-bucket serialization, executor release/recreation, or external executor ownership. The existing targeted compaction/write classes pass locally (29 tests run, 1 skipped), but they exercise mostly the default path. Please add focused behavioral tests before production merge.
3. Please document the thread/memory trade-off and monitoring behavior described in [Feature] Configurable multi-thread async compaction per Flink write subtask (compaction.task-threads) #10132; compactionThreadBusy can exceed 100 with parallel workers. Also reject unsupported values such as 0 and -2 rather than silently treating them as SINGLE.

The feature has clear value, but the current head is not ready to merge.

[JingsongLi]

Reviewed head 072f895 for production use. The end-to-end value is clear: allowing buckets within one long-lived Flink write subtask to compact concurrently can reduce the MOW/DV visibility lag described in #10132. The update addresses the previous feedback on invalid values, documentation, focused executor tests, and Spotless. Local JDK 8 verification passed: the three new test classes 10/10; existing MergeTreeCompactManagerTest, KeyValueFileStoreWriteTest, and BucketedAppendFileStoreWriteTest 26/26; and ordinary mvn -pl paimon-api,paimon-core -DskipTests compile including Spotless. Current full CI is still running.

P1 — retire compaction timers when PER_BUCKET workers are retired. AbstractFileStoreWrite.releaseCompactionExecutor shuts down and removes each bucket executor, but CompactionMetrics.ReporterImpl.getCompactTimer() caches a CompactTimer by thread ID in compactTimers and never removes that entry; ReporterImpl.unregister() removes only the reporter. In a long-lived Flink subtask with changing active partitions/buckets, each replacement worker has a new thread ID. The compactionThreadBusy gauge scans every historical timer, so both retained state and scrape cost grow with total buckets ever compacted, even after their writers are gone. I reproduced this with 32 sequential single-thread workers: after shutting each down and unregistering its reporter, the timer map still held all 32 entries (expected 0). The temporary probe was removed from the review checkout. Please bound or retire these timers on worker shutdown and add a churn regression test for PER_BUCKET; the current routing tests only check executor removal.

This is a feature-specific blocker for the new -1 mode. The default single-thread path and fixed-size pool do not create the same unbounded worker churn.

[jacklong319]

Reviewed head 072f895 for production use. The end-to-end value is clear: allowing buckets within one long-lived Flink write subtask to compact concurrently can reduce the MOW/DV visibility lag described in #10132. The update addresses the previous feedback on invalid values, documentation, focused executor tests, and Spotless. Local JDK 8 verification passed: the three new test classes 10/10; existing MergeTreeCompactManagerTest, KeyValueFileStoreWriteTest, and BucketedAppendFileStoreWriteTest 26/26; and ordinary mvn -pl paimon-api,paimon-core -DskipTests compile including Spotless. Current full CI is still running.

P1 — retire compaction timers when PER_BUCKET workers are retired. AbstractFileStoreWrite.releaseCompactionExecutor shuts down and removes each bucket executor, but CompactionMetrics.ReporterImpl.getCompactTimer() caches a CompactTimer by thread ID in compactTimers and never removes that entry; ReporterImpl.unregister() removes only the reporter. In a long-lived Flink subtask with changing active partitions/buckets, each replacement worker has a new thread ID. The compactionThreadBusy gauge scans every historical timer, so both retained state and scrape cost grow with total buckets ever compacted, even after their writers are gone. I reproduced this with 32 sequential single-thread workers: after shutting each down and unregistering its reporter, the timer map still held all 32 entries (expected 0). The temporary probe was removed from the review checkout. Please bound or retire these timers on worker shutdown and add a churn regression test for PER_BUCKET; the current routing tests only check executor removal.

This is a feature-specific blocker for the new -1 mode. The default single-thread path and fixed-size pool do not create the same unbounded worker churn.

Thanks @JingsongLi — agreed on the P1 for PER_BUCKET timer retention. I’ll fix timer retirement on worker shutdown, add the churn regression test, and update the PR soon.