This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 020a6358757b8043ad84ca7ac9e88bea74536305 Author: Danny Chan <[email protected]> AuthorDate: Wed Sep 23 10:29:24 2026 +0800 perf(table-services): bound compaction planning parallelism (#20024) (cherry picked from commit 6649b5e38a1df194c20b87ff81d9631373f2d6a2) --- .../apache/hudi/config/HoodieCompactionConfig.java | 14 ++++ .../org/apache/hudi/config/HoodieWriteConfig.java | 7 ++ .../BaseHoodieCompactionPlanGenerator.java | 3 +- .../apache/hudi/config/TestHoodieWriteConfig.java | 22 +++++ .../TestBaseHoodieCompactionPlanGenerator.java | 96 ++++++++++++++++++++++ 5 files changed, 141 insertions(+), 1 deletion(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 79b9c962ebda..fbb596c98f7b 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -47,6 +47,15 @@ import static org.apache.hudi.common.config.HoodieReaderConfig.ENABLE_OPTIMIZED_ + "(merging of log files onto a new base files).") public class HoodieCompactionConfig extends HoodieConfig { + public static final ConfigProperty<Integer> COMPACTION_PLAN_PARALLELISM = ConfigProperty + .key("hoodie.compaction.plan.parallelism") + .defaultValue(200) + .markAdvanced() + .sinceVersion("1.3.0") + .withDocumentation("Maximum parallelism for scanning partitions when planning compaction or log compaction. " + + "The actual parallelism is capped by the number of partitions selected by the compaction strategy. " + + "Lower this value to reduce concurrent requests and memory pressure on the file system view server. Must be positive."); + public static final ConfigProperty<String> INLINE_COMPACT = ConfigProperty .key("hoodie.compact.inline") .defaultValue("false") @@ -374,6 +383,11 @@ public class HoodieCompactionConfig extends HoodieConfig { return this; } + public Builder withCompactionPlanParallelism(int parallelism) { + compactionConfig.setValue(COMPACTION_PLAN_PARALLELISM, String.valueOf(parallelism)); + return this; + } + public Builder withScheduleInlineCompaction(Boolean scheduleAsyncCompaction) { compactionConfig.setValue(SCHEDULE_INLINE_COMPACT, String.valueOf(scheduleAsyncCompaction)); return this; diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 0a74fcfd0f12..a3d82f30f6d4 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -1827,6 +1827,10 @@ public class HoodieWriteConfig extends HoodieConfig { return getBoolean(HoodieCompactionConfig.INLINE_COMPACT); } + public int getCompactionPlanParallelism() { + return getInt(HoodieCompactionConfig.COMPACTION_PLAN_PARALLELISM); + } + public boolean scheduleInlineCompaction() { return getBoolean(HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT); } @@ -3707,6 +3711,9 @@ public class HoodieWriteConfig extends HoodieConfig { } } + checkArgument(writeConfig.getCompactionPlanParallelism() > 0, + HoodieCompactionConfig.COMPACTION_PLAN_PARALLELISM.key() + " must be positive"); + boolean inlineCompact = writeConfig.getBoolean(HoodieCompactionConfig.INLINE_COMPACT); boolean inlineCompactSchedule = writeConfig.getBoolean(HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT); checkArgument(!(inlineCompact && inlineCompactSchedule), String.format("Either of inline compaction (%s) or " diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java index daa9d3d5684e..e4607b3e9c26 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java @@ -128,6 +128,7 @@ public abstract class BaseHoodieCompactionPlanGenerator<T extends HoodieRecordPa LOG.info("Last completed instant time {} for table {}", lastCompletedInstantTime, hoodieTable.getConfig().getBasePath()); Option<InstantRange> instantRange = CompactHelpers.getInstance().getInstantRange(metaClient); + int parallelism = Math.min(partitionPaths.size(), writeConfig.getCompactionPlanParallelism()); List<HoodieCompactionOperation> operations = engineContext.flatMap(partitionPaths, partitionPath -> fileSystemView .getLatestFileSlicesStateless(partitionPath) .filter(slice -> filterFileSlice(slice, lastCompletedInstantTime, fgIdsInPendingCompactionAndClustering, instantRange)) @@ -157,7 +158,7 @@ public abstract class BaseHoodieCompactionPlanGenerator<T extends HoodieRecordPa Option<HoodieBaseFile> dataFile = s.getBaseFile(); return new CompactionOperation(dataFile, partitionPath, logFiles, writeConfig.getCompactionStrategy().captureMetrics(writeConfig, s)); - }), partitionPaths.size()).stream() + }), parallelism).stream() .map(CompactionUtils::buildHoodieCompactionOperation).collect(toList()); LOG.info("Total of {} compaction operations are retrieved for table {}", operations.size(), hoodieTable.getConfig().getBasePath()); diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java index 690e8408a2c6..4df39962fb63 100644 --- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/config/TestHoodieWriteConfig.java @@ -49,6 +49,7 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.util.Arrays; +import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.Map; @@ -64,6 +65,27 @@ import static org.junit.jupiter.api.Assertions.assertTrue; public class TestHoodieWriteConfig { + @Test + public void testCompactionPlanParallelism() { + assertEquals(200, HoodieWriteConfig.newBuilder().withPath("/tmp").build().getCompactionPlanParallelism()); + assertEquals(8, HoodieWriteConfig.newBuilder().withPath("/tmp") + .withCompactionConfig(HoodieCompactionConfig.newBuilder().withCompactionPlanParallelism(8).build()) + .build().getCompactionPlanParallelism()); + assertEquals(16, HoodieWriteConfig.newBuilder().withPath("/tmp") + .withProps(Collections.singletonMap(HoodieCompactionConfig.COMPACTION_PLAN_PARALLELISM.key(), "16")) + .build().getCompactionPlanParallelism()); + } + + @ParameterizedTest + @ValueSource(ints = {0, -1}) + public void testCompactionPlanParallelismMustBePositive(int parallelism) { + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> HoodieWriteConfig.newBuilder().withPath("/tmp") + .withProps(Collections.singletonMap(HoodieCompactionConfig.COMPACTION_PLAN_PARALLELISM.key(), String.valueOf(parallelism))) + .build()); + assertTrue(exception.getMessage().contains(HoodieCompactionConfig.COMPACTION_PLAN_PARALLELISM.key())); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) public void testPropertyLoading(boolean withAlternative) throws IOException { diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/compact/plan/generators/TestBaseHoodieCompactionPlanGenerator.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/compact/plan/generators/TestBaseHoodieCompactionPlanGenerator.java new file mode 100644 index 000000000000..12d938ac8efe --- /dev/null +++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/compact/plan/generators/TestBaseHoodieCompactionPlanGenerator.java @@ -0,0 +1,96 @@ +/* + * 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.hudi.table.action.compact.plan.generators; + +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.table.view.SyncableFileSystemView; +import org.apache.hudi.common.testutils.HoodieCommonTestHarness; +import org.apache.hudi.common.testutils.HoodieTestTable; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.config.HoodieCompactionConfig; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.table.HoodieTable; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import java.util.stream.Stream; + +import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getStorageConf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +class TestBaseHoodieCompactionPlanGenerator extends HoodieCommonTestHarness { + + @ParameterizedTest + @CsvSource({"5,2,5,2", "5,10,5,5", "250,200,250,200", "5,10,2,2", "5,2,0,0"}) + void testPlanningParallelism(int partitionCount, int configuredParallelism, int selectedCount, int expectedParallelism) throws Exception { + initMetaClient(); + HoodieTestTable.of(metaClient).addCommit("001"); + metaClient.reloadActiveTimeline(); + HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(basePath) + .withCompactionConfig(HoodieCompactionConfig.newBuilder().withCompactionPlanParallelism(configuredParallelism).build()) + .build(); + HoodieEngineContext context = spy(new HoodieLocalEngineContext(getStorageConf())); + HoodieTable table = mock(HoodieTable.class); + when(table.getMetaClient()).thenReturn(metaClient); + when(table.getConfig()).thenReturn(config); + SyncableFileSystemView view = mock(SyncableFileSystemView.class); + List<String> partitions = IntStream.range(0, partitionCount).mapToObj(i -> "partition-" + i).collect(Collectors.toList()); + List<String> selectedPartitions = partitions.subList(0, selectedCount); + HoodieCompactionPlanGenerator generator = spy(new HoodieCompactionPlanGenerator(table, context, config, null)); + doReturn(partitions).when(generator).getPartitions(); + doReturn(Pair.of(selectedPartitions, partitions.subList(selectedCount, partitionCount))) + .when(generator).filterPartitionPathsByStrategy(partitions); + if (selectedCount > 0) { + when(table.getSliceView()).thenReturn(view); + when(view.getPendingCompactionOperations()).thenReturn(Stream.empty()); + when(view.getFileGroupsInPendingClustering()).thenReturn(Stream.empty()); + for (String partition : selectedPartitions) { + when(view.getLatestFileSlicesStateless(partition)).thenReturn(Stream.empty()); + } + } + + assertNull(generator.generateCompactionPlan("002")); + + if (selectedCount == 0) { + verify(context, never()).flatMap(anyList(), any(), anyInt()); + verifyNoInteractions(view); + } else { + verify(context).flatMap(eq(selectedPartitions), any(), eq(expectedParallelism)); + for (String partition : selectedPartitions) { + verify(view).getLatestFileSlicesStateless(partition); + } + } + } +}
