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

docs/docs/maintenance/metrics.md

@@ -444,7 +444,7 @@ Lookup metrics are available for local partial lookup. They are reported at look
         <tr>
             <td>compactionThreadBusy</td>
             <td>Gauge</td>
-            <td>The maximum business of compaction threads in this task. Currently, there is only one compaction thread in each parallelism, so value of business ranges from 0 (idle) to 100 (compaction running all the time).</td>
+            <td>The maximum busyness of compaction threads in this task, ranging from 0 (idle) to 100 when a single compaction thread is busy all the time. When <code>compaction.task-threads</code> is greater than 1 or set to <code>-1</code> (per-bucket executors), multiple workers may be busy concurrently, so this value can exceed 100.</td>
         </tr>
         <tr>
             <td>avgCompactionTime</td>

docs/docs/primary-key-table/compaction.md

@@ -70,6 +70,29 @@ publishes it. MOW batch readers can opt into merging pending data with
 For `changelog-producer = lookup`, generated changelogs are also delayed. A compactor that cannot
 keep up with sustained input will keep falling behind; relaxing waits does not add capacity.
 
+## Multi-thread async compaction
+
+When many buckets are assigned to the same write task (for example, one Flink sink subtask), the
+default async compaction uses **one shared thread** for all buckets. Under high write throughput,
+compaction may fall behind and level-0 files accumulate. This is especially visible in
+[MOW / deletion vectors mode](./table-mode#merge-on-write), where level-0 data becomes readable
+only after compaction publishes it.
+
+You can increase cross-bucket compaction parallelism with `compaction.task-threads`:
+
+- `1` (default): unchanged — one compaction thread per write task.
+- `N` (`N > 1`): a fixed thread pool of `N` threads shared by all buckets in the task.
+- `-1`: one dedicated compaction thread per active `(partition, bucket)` writer (highest
+  parallelism and memory use).
+
+Compaction **within the same bucket is always serialized**. Values `0` and negative integers other
+than `-1` are rejected.
+
+**Trade-offs:** more compaction threads increase TaskManager memory and I/O concurrency. Size
+TaskManager memory accordingly and monitor [compaction metrics](../maintenance/metrics#compaction-metrics)
+such as `avgLevel0FileCount`, `avgCompactionTime`, and `compactionQueuedCount`. Start with
+`N = 2` or `3` before using `-1`.
+
 ## Dedicated compaction job
 
 Set `write-only = true` on ingest writers and run a

docs/docs/primary-key-table/table-mode.md

@@ -108,6 +108,10 @@ By default, batch reads skip Level-0 files until lookup compaction publishes the
 for this compaction by default. Asynchronous compaction or a dedicated compaction job can delay
 visibility; see [Asynchronous Compaction](./compaction#asynchronous-compaction).
 
+When async compaction falls behind or data visibility latency is high, consider increasing
+`compaction.task-threads` to reduce visibility delay. See
+[Multi-thread async compaction](./compaction#multi-thread-async-compaction).
+
 For batch scans, `deletion-vectors.merge-on-read = true` includes uncompacted data by merging it
 at read time, with additional read cost. It does not change streaming changelog behavior.
 

docs/generated/core_configuration.html

@@ -464,6 +464,12 @@
             <td>MemorySize</td>
             <td>When total size is smaller than this threshold, force a full compaction.</td>
         </tr>
+        <tr>
+            <td><h5>compaction.task-threads</h5></td>
+            <td style="word-wrap: break-word;">1</td>
+            <td>Integer</td>
+            <td>Number of threads for async compaction in each write task (for example, each Flink sink subtask). 1 (default): all buckets in the task share one compaction thread. -1: one dedicated compaction thread per active (partition, bucket) writer in the task so different buckets compact in parallel. N (&gt;1): a fixed thread pool of N threads shared by all buckets in the task. Compaction within the same bucket is still serialized. Values 0 and other negative integers except -1 are not allowed. Higher thread counts increase TaskManager memory pressure; monitor level-0 file count and compaction metrics.</td>
+        </tr>
         <tr>
             <td><h5>consumer-id</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CompactionTaskExecutorMode.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon;
+
+/** Async compaction executor strategy for each write task (for example, a Flink sink subtask). */
+public enum CompactionTaskExecutorMode {
+
+    /** One shared compaction thread serializes compaction for all buckets in the task. */
+    SINGLE,
+
+    /** A fixed-size thread pool ({@code compaction.task-threads}) shared by all buckets. */
+    FIXED_POOL,
+
+    /** One dedicated compaction thread per active (partition, bucket) writer. */
+    PER_BUCKET
+}

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -965,6 +965,22 @@ public InlineElement getDescription() {
                     .defaultValue(false)
                     .withDescription("Whether to force a compaction before commit.");
 
+    public static final ConfigOption<Integer> COMPACTION_TASK_THREADS =
+            key("compaction.task-threads")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription(
+                            "Number of threads for async compaction in each write task (for example, "
+                                    + "each Flink sink subtask). "
+                                    + "1 (default): all buckets in the task share one compaction thread. "
+                                    + "-1: one dedicated compaction thread per active (partition, bucket) "
+                                    + "writer in the task so different buckets compact in parallel. "
+                                    + "N (>1): a fixed thread pool of N threads shared by all buckets in the task. "
+                                    + "Compaction within the same bucket is still serialized. "
+                                    + "Values 0 and other negative integers except -1 are not allowed. "
+                                    + "Higher thread counts increase TaskManager memory pressure; "
+                                    + "monitor level-0 file count and compaction metrics.");
+
     public static final ConfigOption<SequenceNumberInitMode> WRITE_SEQUENCE_NUMBER_INIT_MODE =
             key("write.sequence-number-init-mode")
                     .enumType(SequenceNumberInitMode.class)
@@ -3888,6 +3904,26 @@ public boolean commitForceCompact() {
         return options.get(COMMIT_FORCE_COMPACT);
     }
 
+    public CompactionTaskExecutorMode compactionTaskExecutorMode() {
+        int threads = compactionTaskThreads();
+        if (threads == -1) {
+            return CompactionTaskExecutorMode.PER_BUCKET;
+        }
+        if (threads == 1) {
+            return CompactionTaskExecutorMode.SINGLE;
+        }
+        return CompactionTaskExecutorMode.FIXED_POOL;
+    }
+
+    public int compactionTaskThreads() {
+        int threads = options.get(COMPACTION_TASK_THREADS);
+        checkArgument(
+                threads == -1 || threads > 0,
+                "The option %s must be -1, 1, or any integer greater than 1.",
+                COMPACTION_TASK_THREADS.key());
+        return threads;
+    }
+
     public SequenceNumberInitMode writeSequenceNumberInitMode() {
         return options.get(WRITE_SEQUENCE_NUMBER_INIT_MODE);
     }

paimon-api/src/test/java/org/apache/paimon/CoreOptionsCompactionTaskThreadsTest.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon;
+
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link CoreOptions#COMPACTION_TASK_THREADS}. */
+class CoreOptionsCompactionTaskThreadsTest {
+
+    @Test
+    void testDefaultIsSingleThreadMode() {
+        CoreOptions options = new CoreOptions(new Options());
+        assertThat(options.compactionTaskExecutorMode())
+                .isEqualTo(CompactionTaskExecutorMode.SINGLE);
+        assertThat(options.compactionTaskThreads()).isEqualTo(1);
+    }
+
+    @Test
+    void testFixedPoolMode() {
+        Options options = new Options();
+        options.set(CoreOptions.COMPACTION_TASK_THREADS, 3);
+        CoreOptions coreOptions = new CoreOptions(options);
+        assertThat(coreOptions.compactionTaskExecutorMode())
+                .isEqualTo(CompactionTaskExecutorMode.FIXED_POOL);
+        assertThat(coreOptions.compactionTaskThreads()).isEqualTo(3);
+    }
+
+    @Test
+    void testPerBucketMode() {
+        Options options = new Options();
+        options.set(CoreOptions.COMPACTION_TASK_THREADS, -1);
+        CoreOptions coreOptions = new CoreOptions(options);
+        assertThat(coreOptions.compactionTaskExecutorMode())
+                .isEqualTo(CompactionTaskExecutorMode.PER_BUCKET);
+    }
+
+    @Test
+    void testRejectZeroAndOtherNegativeValues() {
+        assertInvalid(0);
+        assertInvalid(-2);
+        assertInvalid(-100);
+    }
+
+    private static void assertInvalid(int threads) {
+        Options options = new Options();
+        options.set(CoreOptions.COMPACTION_TASK_THREADS, threads);
+        CoreOptions coreOptions = new CoreOptions(options);
+        assertThatThrownBy(coreOptions::compactionTaskExecutorMode)
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining(CoreOptions.COMPACTION_TASK_THREADS.key());
+    }
+}