diff --git a/docs/docs/maintenance/metrics.md b/docs/docs/maintenance/metrics.md index efd78e520d16..c2f18f0a8d79 100644 --- a/docs/docs/maintenance/metrics.md +++ b/docs/docs/maintenance/metrics.md @@ -444,7 +444,7 @@ Lookup metrics are available for local partial lookup. They are reported at look compactionThreadBusy Gauge - 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). + 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 compaction.task-threads is greater than 1 or set to -1 (per-bucket executors), multiple workers may be busy concurrently, so this value can exceed 100. avgCompactionTime diff --git a/docs/docs/primary-key-table/compaction.md b/docs/docs/primary-key-table/compaction.md index f69c587aa290..ae0a97360e65 100644 --- a/docs/docs/primary-key-table/compaction.md +++ b/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 diff --git a/docs/docs/primary-key-table/table-mode.md b/docs/docs/primary-key-table/table-mode.md index 78bb6eb24234..b18c7dabca28 100644 --- a/docs/docs/primary-key-table/table-mode.md +++ b/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. diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 750f6131fa0d..6aad4d4b08db 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -464,6 +464,12 @@ MemorySize When total size is smaller than this threshold, force a full compaction. + +
compaction.task-threads
+ 1 + Integer + 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. +
consumer-id
(none) diff --git a/paimon-api/src/main/java/org/apache/paimon/CompactionTaskExecutorMode.java b/paimon-api/src/main/java/org/apache/paimon/CompactionTaskExecutorMode.java new file mode 100644 index 000000000000..a4404bd16e85 --- /dev/null +++ b/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 +} diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 9818e68d9ca9..81399cc58c9e 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/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 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 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); } diff --git a/paimon-api/src/test/java/org/apache/paimon/CoreOptionsCompactionTaskThreadsTest.java b/paimon-api/src/test/java/org/apache/paimon/CoreOptionsCompactionTaskThreadsTest.java new file mode 100644 index 000000000000..ea9c48955c05 --- /dev/null +++ b/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()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java index 0c0738c6173e..5d5013d218b6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java @@ -18,6 +18,7 @@ package org.apache.paimon.operation; +import org.apache.paimon.CompactionTaskExecutorMode; import org.apache.paimon.CoreOptions; import org.apache.paimon.KeyValue; import org.apache.paimon.Snapshot; @@ -61,7 +62,9 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.OptionalLong; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Function; @@ -101,6 +104,11 @@ public abstract class AbstractFileStoreWrite implements FileStoreWrite { protected final Map>> writers; protected WriteRestore restore; + private final CompactionTaskExecutorMode compactionTaskExecutorMode; + private final int compactionTaskThreads; + private final Map perBucketCompactExecutors = + new ConcurrentHashMap<>(); + private boolean externalCompactExecutor = false; private ExecutorService lazyCompactExecutor; private ExecutorService lazyPrimaryKeyIndexExecutor; private boolean closeCompactExecutorWhenLeaving = true; @@ -142,6 +150,8 @@ protected AbstractFileStoreWrite( this.writerNumberMax = options.writeMaxWritersToSpill(); this.legacyPartitionName = options.legacyPartitionName(); this.options = options; + this.compactionTaskExecutorMode = options.compactionTaskExecutorMode(); + this.compactionTaskThreads = options.compactionTaskThreads(); this.partitionTimestampValidator = PartitionTimestampValidator.create(options, partitionType); } @@ -182,6 +192,7 @@ public void withIgnoreNumBucketCheck(boolean ignoreNumBucketCheck) { public void withCompactExecutor(ExecutorService compactExecutor) { this.lazyCompactExecutor = compactExecutor; this.closeCompactExecutorWhenLeaving = false; + this.externalCompactExecutor = true; } @Override @@ -313,6 +324,7 @@ public List prepareCommit(boolean waitCompaction, long commitIden if (writerContainer.primaryKeyIndexMaintainer != null) { writerContainer.primaryKeyIndexMaintainer.close(); } + releaseCompactionExecutor(partition, bucket); bucketIter.remove(); } } else { @@ -384,9 +396,7 @@ public void close() throws Exception { // also left both thread pools running for the life of the process. None of the // calls below throws, so the writer failure is never replaced by one of them. writers.clear(); - if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) { - lazyCompactExecutor.shutdownNow(); - } + shutdownCompactionExecutors(); if (lazyPrimaryKeyIndexExecutor != null) { lazyPrimaryKeyIndexExecutor.shutdownNow(); } @@ -458,7 +468,7 @@ public void restore(List> states) { state.dataFiles, state.maxSequenceNumber, state.commitIncrement, - compactExecutor(), + compactExecutor(state.partition, state.bucket), state.deletionVectorsMaintainer, // Restore reconstructs writer state from checkpointed files, so do // not ignore them. @@ -591,7 +601,7 @@ private WriterContainer createWriterContainer( startingMaxSequenceNumber( getMaxSequenceNumber(restoreFiles), latestSnapshot), null, - compactExecutor(), + compactExecutor(partition, bucket), dvMaintainer, actualIgnorePreviousFiles); notifyNewWriter(writer); @@ -703,12 +713,94 @@ private void checkNumBuckets(String partInfo, int expected, int previous) { } } - private ExecutorService compactExecutor() { + private ExecutorService compactExecutor(BinaryRow partition, int bucket) { + if (externalCompactExecutor) { + return lazyCompactExecutor; + } + + switch (compactionTaskExecutorMode) { + case PER_BUCKET: + return perBucketCompactExecutors.computeIfAbsent( + new BucketCompactionExecutorKey(partition, bucket), + key -> + Executors.newSingleThreadExecutor( + new ExecutorThreadFactory( + Thread.currentThread().getName() + + "-compaction-bucket-" + + key.bucket))); + case FIXED_POOL: + return sharedCompactionExecutor(compactionTaskThreads, "-compaction-pool"); + case SINGLE: + default: + return sharedCompactionExecutor(1, "-compaction"); + } + } + + private void releaseCompactionExecutor(BinaryRow partition, int bucket) { + if (compactionTaskExecutorMode != CompactionTaskExecutorMode.PER_BUCKET + || externalCompactExecutor) { + return; + } + + ExecutorService removed = + perBucketCompactExecutors.remove( + new BucketCompactionExecutorKey(partition, bucket)); + if (removed != null) { + removed.shutdownNow(); + } + } + + private void shutdownCompactionExecutors() { + for (ExecutorService executor : perBucketCompactExecutors.values()) { + executor.shutdownNow(); + } + perBucketCompactExecutors.clear(); + + if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) { + lazyCompactExecutor.shutdownNow(); + lazyCompactExecutor = null; + } + } + + private static final class BucketCompactionExecutorKey { + private final BinaryRow partition; + private final int bucket; + + private BucketCompactionExecutorKey(BinaryRow partition, int bucket) { + this.partition = partition; + this.bucket = bucket; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + BucketCompactionExecutorKey that = (BucketCompactionExecutorKey) o; + return bucket == that.bucket && Objects.equals(partition, that.partition); + } + + @Override + public int hashCode() { + return Objects.hash(partition, bucket); + } + } + + private ExecutorService sharedCompactionExecutor(int threads, String nameSuffix) { if (lazyCompactExecutor == null) { - lazyCompactExecutor = - Executors.newSingleThreadScheduledExecutor( - new ExecutorThreadFactory( - Thread.currentThread().getName() + "-compaction")); + String threadNamePrefix = Thread.currentThread().getName() + nameSuffix; + if (threads <= 1) { + lazyCompactExecutor = + Executors.newSingleThreadScheduledExecutor( + new ExecutorThreadFactory(threadNamePrefix)); + } else { + lazyCompactExecutor = + Executors.newFixedThreadPool( + threads, new ExecutorThreadFactory(threadNamePrefix)); + } } return lazyCompactExecutor; } @@ -723,6 +815,21 @@ private ExecutorService primaryKeyIndexExecutor() { return lazyPrimaryKeyIndexExecutor; } + @VisibleForTesting + public ExecutorService compactExecutorForTesting(BinaryRow partition, int bucket) { + return compactExecutor(partition, bucket); + } + + @VisibleForTesting + public int activePerBucketExecutorCountForTesting() { + return perBucketCompactExecutors.size(); + } + + @VisibleForTesting + public void releaseCompactionExecutorForTesting(BinaryRow partition, int bucket) { + releaseCompactionExecutor(partition, bucket); + } + @VisibleForTesting public ExecutorService getCompactExecutor() { return lazyCompactExecutor; diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java index ed6122b2b6dc..b94cd1447bcf 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/metrics/CompactionMetrics.java @@ -25,9 +25,11 @@ import org.apache.paimon.metrics.MetricRegistry; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; import java.util.Objects; import java.util.Queue; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.stream.DoubleStream; @@ -66,6 +68,8 @@ public class CompactionMetrics { private final MetricGroup metricGroup; private final Map reporters; private final Map compactTimers; + private final Map compactTimerRefCounts; + private final Map compactTimerLocks; private final Queue compactionTimes; private Counter compactionsCompletedCounter; private Counter compactionsTotalCounter; @@ -75,6 +79,8 @@ public CompactionMetrics(MetricRegistry registry, String tableName) { this.metricGroup = registry.createTableMetricGroup(GROUP_NAME, tableName); this.reporters = new HashMap<>(); this.compactTimers = new ConcurrentHashMap<>(); + this.compactTimerRefCounts = new ConcurrentHashMap<>(); + this.compactTimerLocks = new ConcurrentHashMap<>(); this.compactionTimes = new ConcurrentLinkedQueue<>(); registerGenericCompactionMetrics(); @@ -85,6 +91,30 @@ public MetricGroup getMetricGroup() { return metricGroup; } + @VisibleForTesting + int activeCompactTimerCount() { + return compactTimers.size(); + } + + private Object compactTimerLock(long threadId) { + return compactTimerLocks.computeIfAbsent(threadId, ignored -> new Object()); + } + + private void releaseCompactTimer(long threadId) { + synchronized (compactTimerLock(threadId)) { + compactTimerRefCounts.compute( + threadId, + (id, count) -> { + if (count == null || count <= 1) { + compactTimers.remove(id); + compactTimerLocks.remove(id); + return null; + } + return count - 1; + }); + } + } + private void registerGenericCompactionMetrics() { metricGroup.gauge(MAX_LEVEL0_FILE_COUNT, () -> getLevel0FileCountStream().max().orElse(-1)); metricGroup.gauge( @@ -218,6 +248,7 @@ private class ReporterImpl implements Reporter { private long totalFileCount = 0; private long sortBufferUsedBytes = 0; private double sortBufferUtilisationPercent = 0.0; + private final Set compactThreadIds = new HashSet<>(); private ReporterImpl(PartitionAndBucket key) { this.key = key; @@ -226,9 +257,16 @@ private ReporterImpl(PartitionAndBucket key) { @Override public CompactTimer getCompactTimer() { - return compactTimers.computeIfAbsent( - Thread.currentThread().getId(), - ignore -> new CompactTimer(BUSY_MEASURE_MILLIS)); + long threadId = Thread.currentThread().getId(); + synchronized (compactTimerLock(threadId)) { + CompactTimer timer = + compactTimers.computeIfAbsent( + threadId, ignore -> new CompactTimer(BUSY_MEASURE_MILLIS)); + if (compactThreadIds.add(threadId)) { + compactTimerRefCounts.merge(threadId, 1, Integer::sum); + } + return timer; + } } @Override @@ -295,6 +333,10 @@ public void decreaseCompactionsQueuedCount() { @Override public void unregister() { + for (Long threadId : compactThreadIds) { + releaseCompactTimer(threadId); + } + compactThreadIds.clear(); reporters.remove(key); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerBucketSerializationTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerBucketSerializationTest.java new file mode 100644 index 000000000000..491a3f905fd6 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerBucketSerializationTest.java @@ -0,0 +1,125 @@ +/* + * 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.mergetree.compact; + +import org.apache.paimon.compact.CompactResult; +import org.apache.paimon.compact.CompactUnit; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataFileTestUtils; +import org.apache.paimon.mergetree.LevelSortedRun; +import org.apache.paimon.mergetree.Levels; +import org.apache.paimon.mergetree.SortedRun; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests that a single {@link MergeTreeCompactManager} serializes compaction per bucket. */ +class MergeTreeCompactManagerBucketSerializationTest { + + private ExecutorService executorService; + + @AfterEach + void tearDown() { + if (executorService != null) { + executorService.shutdownNow(); + executorService = null; + } + } + + @Test + void testSecondTriggerIgnoredWhileCompactionRunning() throws Exception { + executorService = Executors.newSingleThreadExecutor(); + AtomicInteger rewriteCalls = new AtomicInteger(); + CountDownLatch rewriteStarted = new CountDownLatch(1); + CountDownLatch unblockRewrite = new CountDownLatch(1); + + DataFileMeta file1 = DataFileTestUtils.newFile(0, 1, 3, 3L); + DataFileMeta file2 = DataFileTestUtils.newFile(0, 4, 6, 6L); + LevelSortedRun run1 = new LevelSortedRun(0, SortedRun.fromSingle(file1)); + LevelSortedRun run2 = new LevelSortedRun(0, SortedRun.fromSingle(file2)); + List runs = Arrays.asList(run1, run2); + + Levels levels = mock(Levels.class); + when(levels.levelSortedRuns()).thenReturn(runs); + when(levels.numberOfLevels()).thenReturn(3); + when(levels.nonEmptyHighestLevel()).thenReturn(0); + + CompactStrategy strategy = mock(CompactStrategy.class); + when(strategy.pick(anyInt(), any())) + .thenReturn(Optional.of(CompactUnit.fromLevelRuns(1, runs))); + + CompactRewriter rewriter = mock(CompactRewriter.class); + when(rewriter.rewrite(anyInt(), anyBoolean(), any())) + .thenAnswer( + invocation -> { + rewriteCalls.incrementAndGet(); + rewriteStarted.countDown(); + assertThat(unblockRewrite.await(30, TimeUnit.SECONDS)).isTrue(); + return new CompactResult( + Collections.emptyList(), Collections.emptyList()); + }); + + MergeTreeCompactManager manager = + new MergeTreeCompactManager( + executorService, + levels, + strategy, + Comparator.comparingInt(row -> row.getInt(0)), + 1024 * 1024, + 5, + rewriter, + null, + null, + false, + false, + null, + false, + false, + ""); + + manager.triggerCompaction(false); + assertThat(rewriteStarted.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(manager.compactNotCompleted()).isTrue(); + + manager.triggerCompaction(false); + assertThat(rewriteCalls.get()).isEqualTo(1); + + unblockRewrite.countDown(); + manager.getCompactionResult(true); + assertThat(manager.compactNotCompleted()).isFalse(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/CompactionTaskExecutorRoutingTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/CompactionTaskExecutorRoutingTest.java new file mode 100644 index 000000000000..4fe13699ef9a --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/operation/CompactionTaskExecutorRoutingTest.java @@ -0,0 +1,192 @@ +/* + * 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.operation; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.deletionvectors.BucketedDvMaintainer; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.CommitIncrement; +import org.apache.paimon.utils.RecordWriter; +import org.apache.paimon.utils.SnapshotManager; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.function.Function; + +import static org.apache.paimon.data.BinaryRow.EMPTY_ROW; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +/** Focused behavioral tests for {@code compaction.task-threads} executor routing. */ +class CompactionTaskExecutorRoutingTest { + + private ExecutorService externalExecutor; + + @AfterEach + void tearDown() { + if (externalExecutor != null) { + externalExecutor.shutdownNow(); + externalExecutor = null; + } + } + + @Test + void testFixedPoolUsesSharedThreadPoolWithConfiguredSize() throws Exception { + ExecutorRoutingWrite write = new ExecutorRoutingWrite(coreOptions(2)); + ExecutorService bucket0 = write.compactExecutorForTesting(EMPTY_ROW, 0); + ExecutorService bucket1 = write.compactExecutorForTesting(EMPTY_ROW, 1); + assertThat(bucket0).isSameAs(bucket1); + assertThat(bucket0).isInstanceOf(ThreadPoolExecutor.class); + assertThat(((ThreadPoolExecutor) bucket0).getMaximumPoolSize()).isEqualTo(2); + write.close(); + } + + @Test + void testFixedPoolAllowsCrossBucketParallelism() throws Exception { + ExecutorRoutingWrite write = new ExecutorRoutingWrite(coreOptions(2)); + ExecutorService pool = write.compactExecutorForTesting(EMPTY_ROW, 0); + + CountDownLatch firstStarted = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch secondStarted = new CountDownLatch(1); + + Future first = + pool.submit( + () -> { + firstStarted.countDown(); + try { + releaseFirst.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + Future second = + pool.submit( + () -> { + secondStarted.countDown(); + }); + + assertThat(firstStarted.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(secondStarted.await(30, TimeUnit.SECONDS)).isTrue(); + + releaseFirst.countDown(); + first.get(30, TimeUnit.SECONDS); + second.get(30, TimeUnit.SECONDS); + write.close(); + } + + @Test + void testPerBucketUsesDedicatedExecutorsAndReusesSameBucketExecutor() throws Exception { + BinaryRow partition = EMPTY_ROW.copy(); + ExecutorRoutingWrite write = new ExecutorRoutingWrite(coreOptions(-1)); + + ExecutorService bucket0 = write.compactExecutorForTesting(partition, 0); + ExecutorService bucket1 = write.compactExecutorForTesting(partition, 1); + assertThat(bucket0).isNotSameAs(bucket1); + assertThat(write.activePerBucketExecutorCountForTesting()).isEqualTo(2); + + ExecutorService bucket0Again = write.compactExecutorForTesting(partition, 0); + assertThat(bucket0Again).isSameAs(bucket0); + write.close(); + } + + @Test + void testPerBucketReleaseShutsDownAndRecreatesExecutor() throws Exception { + BinaryRow partition = EMPTY_ROW.copy(); + ExecutorRoutingWrite write = new ExecutorRoutingWrite(coreOptions(-1)); + ExecutorService first = write.compactExecutorForTesting(partition, 0); + assertThat(write.activePerBucketExecutorCountForTesting()).isEqualTo(1); + + write.releaseCompactionExecutorForTesting(partition, 0); + assertThat(write.activePerBucketExecutorCountForTesting()).isZero(); + assertThat(first.isShutdown()).isTrue(); + + ExecutorService second = write.compactExecutorForTesting(partition, 0); + assertThat(second).isNotSameAs(first); + write.close(); + } + + @Test + void testExternalExecutorIsSharedAndNotClosedByWrite() throws Exception { + externalExecutor = Executors.newSingleThreadExecutor(); + ExecutorRoutingWrite write = new ExecutorRoutingWrite(coreOptions(-1)); + write.withCompactExecutor(externalExecutor); + + ExecutorService bucket0 = write.compactExecutorForTesting(EMPTY_ROW, 0); + ExecutorService bucket1 = write.compactExecutorForTesting(EMPTY_ROW, 1); + assertThat(bucket0).isSameAs(externalExecutor); + assertThat(bucket1).isSameAs(externalExecutor); + assertThat(write.activePerBucketExecutorCountForTesting()).isZero(); + + write.close(); + assertThat(externalExecutor.isShutdown()).isFalse(); + } + + private static CoreOptions coreOptions(int compactionTaskThreads) { + Options options = new Options(); + options.set(CoreOptions.COMPACTION_TASK_THREADS, compactionTaskThreads); + return new CoreOptions(options); + } + + private static class ExecutorRoutingWrite extends AbstractFileStoreWrite { + + private ExecutorRoutingWrite(CoreOptions coreOptions) { + super( + mock(SnapshotManager.class), + mock(FileStoreScan.class), + null, + null, + null, + "test-table", + coreOptions, + RowType.of()); + } + + @Override + protected Function, Boolean> createWriterCleanChecker() { + return writer -> false; + } + + @Override + protected RecordWriter createWriter( + BinaryRow partition, + int bucket, + List restoreFiles, + long restoredMaxSeqNumber, + @Nullable CommitIncrement restoreIncrement, + ExecutorService compactExecutor, + @Nullable BucketedDvMaintainer deletionVectorsMaintainer, + boolean ignorePreviousFiles) { + return mock(RecordWriter.class); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java index 380eac4b918a..afe980a88023 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/metrics/CompactionMetricsTest.java @@ -48,7 +48,13 @@ import java.util.Arrays; import java.util.Collections; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; @@ -257,6 +263,91 @@ public void testTotalFileSizeForPrimaryKeyTables() throws Exception { commit.close(); } + @Test + public void testCompactTimersRetiredAfterPerBucketWorkerChurn() throws Exception { + CompactionMetrics metrics = new CompactionMetrics(new TestMetricRegistry(), "myTable"); + for (int i = 0; i < 32; i++) { + ExecutorService worker = Executors.newSingleThreadExecutor(); + CompactionMetrics.Reporter reporter = metrics.createReporter(BinaryRow.EMPTY_ROW, i); + try { + worker.submit( + () -> { + reporter.getCompactTimer().start(); + reporter.getCompactTimer().finish(); + }) + .get(30, TimeUnit.SECONDS); + } finally { + reporter.unregister(); + worker.shutdownNow(); + } + } + assertThat(metrics.activeCompactTimerCount()).isZero(); + } + + @Test + public void testCompactTimerConcurrentUnregisterAndStartOnSharedWorker() throws Exception { + for (int attempt = 0; attempt < 200; attempt++) { + CompactionMetrics metrics = new CompactionMetrics(new TestMetricRegistry(), "myTable"); + ExecutorService worker = Executors.newSingleThreadExecutor(); + CompactionMetrics.Reporter retiring = metrics.createReporter(BinaryRow.EMPTY_ROW, 0); + CompactionMetrics.Reporter starting = metrics.createReporter(BinaryRow.EMPTY_ROW, 1); + CountDownLatch compactionStarted = new CountDownLatch(1); + CountDownLatch allowWorkerContinue = new CountDownLatch(1); + AtomicReference workerError = new AtomicReference<>(); + + Future compaction = + worker.submit( + () -> { + try { + retiring.getCompactTimer().start(); + retiring.getCompactTimer().finish(); + compactionStarted.countDown(); + allowWorkerContinue.await(30, TimeUnit.SECONDS); + starting.getCompactTimer().start(); + starting.getCompactTimer().finish(); + } catch (Throwable t) { + workerError.set(t); + } + }); + + assertThat(compactionStarted.await(30, TimeUnit.SECONDS)).isTrue(); + retiring.unregister(); + allowWorkerContinue.countDown(); + + compaction.get(30, TimeUnit.SECONDS); + worker.shutdownNow(); + + assertThat(workerError.get()).isNull(); + starting.unregister(); + assertThat(metrics.activeCompactTimerCount()).isZero(); + } + } + + @Test + public void testCompactTimerKeptWhileSharedCompactionThreadInUse() throws Exception { + CompactionMetrics metrics = new CompactionMetrics(new TestMetricRegistry(), "myTable"); + ExecutorService sharedPool = Executors.newFixedThreadPool(1); + CompactionMetrics.Reporter first = metrics.createReporter(BinaryRow.EMPTY_ROW, 0); + CompactionMetrics.Reporter second = metrics.createReporter(BinaryRow.EMPTY_ROW, 1); + try { + sharedPool + .submit( + () -> { + first.getCompactTimer().start(); + first.getCompactTimer().finish(); + second.getCompactTimer().start(); + second.getCompactTimer().finish(); + }) + .get(30, TimeUnit.SECONDS); + first.unregister(); + assertThat(metrics.activeCompactTimerCount()).isEqualTo(1); + second.unregister(); + assertThat(metrics.activeCompactTimerCount()).isZero(); + } finally { + sharedPool.shutdownNow(); + } + } + private Object getMetric(CompactionMetrics metrics, String metricName) { Metric metric = metrics.getMetricGroup().getMetrics().get(metricName); if (metric instanceof Gauge) {