[Feature] Configurable multi-thread async compaction per Flink write subtask (compaction.task-threads)

[jacklong319]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Problem we hit in production

We run large Flink ingest jobs into Paimon primary-key tables with deletion-vectors.enabled=true (MOW + DV for fast reads). Each Flink write subtask only has one shared async compaction thread for every (partition, bucket) writer on that subtask.

During overnight / peak ETL windows, compaction cannot keep up with write throughput:

• L0 file count spikes (avgLevel0FileCount stays high in peak).
• Compaction duration grows (avgCompactionTime often reaches several minutes).
• The compaction queue does not drain fast enough (compactionQueuedCount stays elevated).

In Deletion Vectors mode, Paimon documents that L0 data is only visible to readers after compaction (with async compaction, readers naturally lag writers). When many buckets on one subtask are compacted one after another on a single thread, buckets progress at different speeds, backlog grows, and downstream jobs that expect “current” data read incomplete snapshots. In our workload we compared row counts against a Doris table fed from the same upstream and saw up to ~40 minutes maximum lag with the default single-thread compaction.

That breaks timeliness-sensitive use cases: scheduled ETL, reconciliation, and monitoring that must read complete Paimon data while still benefiting from DV scan performance.

Why existing options were not enough for us

• Tuning write/compaction triggers alone did not remove the serial cross-bucket bottleneck inside one subtask when many buckets are assigned to the same task.
• Dedicated compaction is valuable but we also need inline write-path async compaction to keep up during peak without operating a separate job for every table.
• There is no table option today to increase parallelism of async compaction across buckets within the same write subtask while keeping per-bucket compaction serialized (required for correctness).

How this feature fixes it

We propose compaction.task-threads: configurable parallel async compaction inside each write subtask, with unchanged serialization per bucket (safe for LSM invariants). Default 1 preserves current community behavior; N > 1 uses a fixed pool shared by buckets; -1 optionally dedicates one thread per active bucket (higher memory risk).

After enabling multiple threads in production (ZTO):

• L0 curves in peak became flatter; peak L0 count dropped roughly in line with compaction capacity.
• Read delay vs Doris improved from ~40 min (1 thread) → ~20 min (2 threads) → ~8 min (3 threads) with the same write load.
• Internal reconciliation tests showed no data accuracy regression vs single-thread baseline.

Trade-off: more compaction threads increase TaskManager memory pressure; TM heap must be sized accordingly (especially for -1 / per-bucket mode).

Motivation summary

We need a first-class, backward-compatible knob for cross-bucket async compaction parallelism per subtask so that DV mode can deliver both good query performance and acceptable freshness under high bucket count and peak write load—without changing per-bucket correctness guarantees.

Solution

1. Upstream compaction.task-threads in CoreOptions (already defined in our branch) with semantics above.
2. Introduce CompactionTaskExecutorMode and route async compaction executors in AbstractFileStoreWrite (per write subtask / sink task).
   Optionally mention withCompactExecutor for tests or external injection if you expose that.
3. Documentation:
   • primary-key-table/compaction.md — new section “Multi-thread async compaction”
   • primary-key-table/table-mode.md — cross-link DV visibility + compaction.task-threads
   • maintenance/metrics.md — note compactionThreadBusy can exceed 100 when multiple threads are busy
4. Guidance: recommend starting with N = 2 or 3, monitor L0 + TM memory; avoid -1 unless understood.

Anything else?

My understanding:

• Community already documents async compaction and DV visibility rules; the missing knob is parallelism of compaction work across buckets inside one subtask.
• Serial single-thread compaction is a throughput bottleneck for wide bucket assignment, not a correctness requirement.

Design principles:

• Per-bucket serialization unchanged (one compact stream per bucket).
• Cross-bucket parallelism via thread pool or per-bucket executors.
• Default 1 preserves today’s behavior.

POC / production:

• Validated on large partitioned tables at ZTO with real overnight ETL and Doris parity checks.
• Willing to add unit/integration tests and doc PR once issue is agreed.

Are you willing to submit a PR?

• I'm willing to submit a PR!