jacklong319 opened a new issue, #10132: URL: https://github.com/apache/paimon/issues/10132
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/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](https://paimon.apache.org/docs/master/maintenance/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? - [x] I'm willing to submit a PR! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
