From 2a0834fee712c0e1c2d959ef70c4a2ac291eec6d Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Fri, 14 Aug 2026 16:24:36 +0800 Subject: [PATCH 1/3] [spark][flink] Support bucket-level compaction for fixed-bucket tables --- .../apache/paimon/utils/ParameterUtils.java | 40 ++++++++++++ .../paimon/utils/ParameterUtilsTest.java | 55 ++++++++++++++++ .../flink/procedure/CompactProcedure.java | 29 +++++++++ .../paimon/flink/action/CompactAction.java | 51 +++++++++++++++ .../flink/action/CompactActionFactory.java | 12 +++- .../flink/action/SortCompactAction.java | 4 ++ .../flink/procedure/CompactProcedure.java | 11 +++- .../flink/source/CompactorSourceBuilder.java | 10 +++ .../flink/action/CompactActionITCase.java | 64 +++++++++++++++++++ .../procedure/CompactProcedureITCase.java | 40 ++++++++++++ .../spark/procedure/CompactProcedure.java | 40 +++++++++++- .../procedure/CompactProcedureTestBase.scala | 39 +++++++++++ 12 files changed, 390 insertions(+), 5 deletions(-) create mode 100644 paimon-common/src/test/java/org/apache/paimon/utils/ParameterUtilsTest.java diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java index 2eaa4ba94c86..d934a9ae6aa3 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java @@ -30,12 +30,52 @@ import java.util.ArrayList; import java.util.HashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; /** This is a util class for converting string parameter to another format. */ public class ParameterUtils { + private static final Pattern INTEGER_RANGE = + Pattern.compile("([0-9]+)(?:\\s*-\\s*([0-9]+))?"); + + public static List parseIntegerRanges(String values, int exclusiveUpperBound) { + Preconditions.checkArgument( + !StringUtils.isNullOrWhitespaceOnly(values), + "Integer ranges must not be empty."); + Preconditions.checkArgument( + exclusiveUpperBound > 0, "Exclusive upper bound must be greater than 0."); + Set result = new LinkedHashSet<>(); + for (String token : values.split(",", -1)) { + String trimmedToken = token.trim(); + Preconditions.checkArgument( + !trimmedToken.isEmpty(), "Integer ranges must not contain an empty item."); + Matcher matcher = INTEGER_RANGE.matcher(trimmedToken); + Preconditions.checkArgument( + matcher.matches(), "Invalid integer or range: '%s'.", trimmedToken); + long start = Long.parseLong(matcher.group(1)); + long end = matcher.group(2) == null ? start : Long.parseLong(matcher.group(2)); + Preconditions.checkArgument( + start <= end, + "Integer range start %s must not be greater than end %s.", + start, + end); + Preconditions.checkArgument( + end < exclusiveUpperBound, + "Integer or range '%s' is out of range [0, %s).", + trimmedToken, + exclusiveUpperBound); + for (long value = start; value <= end; value++) { + result.add((int) value); + } + } + return new ArrayList<>(result); + } + public static List> getPartitions(String... partitionStrings) { List> partitions = new ArrayList<>(); for (String partition : partitionStrings) { diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/ParameterUtilsTest.java b/paimon-common/src/test/java/org/apache/paimon/utils/ParameterUtilsTest.java new file mode 100644 index 000000000000..47f1be885f50 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/utils/ParameterUtilsTest.java @@ -0,0 +1,55 @@ +/* + * 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.utils; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link ParameterUtils}. */ +class ParameterUtilsTest { + + @Test + void testParseIntegerRanges() { + assertThat(ParameterUtils.parseIntegerRanges("0-2, 4, 2, 6 - 7", 8)) + .isEqualTo(Arrays.asList(0, 1, 2, 4, 6, 7)); + } + + @Test + void testInvalidIntegerRanges() { + assertThatThrownBy(() -> ParameterUtils.parseIntegerRanges("", 8)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must not be empty"); + assertThatThrownBy(() -> ParameterUtils.parseIntegerRanges("0,,2", 8)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("empty item"); + assertThatThrownBy(() -> ParameterUtils.parseIntegerRanges("3-1", 8)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must not be greater"); + assertThatThrownBy(() -> ParameterUtils.parseIntegerRanges("-1", 8)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Invalid integer or range"); + assertThatThrownBy(() -> ParameterUtils.parseIntegerRanges("0-8", 8)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("out of range"); + } +} diff --git a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java index 18e03e053cdd..52e4d4ba2cc2 100644 --- a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java +++ b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java @@ -139,6 +139,31 @@ public String[] call( String partitionIdleTime, String compactStrategy) throws Exception { + return call( + procedureContext, + tableId, + partitions, + orderStrategy, + orderByColumns, + tableOptions, + whereSql, + partitionIdleTime, + compactStrategy, + null); + } + + public String[] call( + ProcedureContext procedureContext, + String tableId, + String partitions, + String orderStrategy, + String orderByColumns, + String tableOptions, + String whereSql, + String partitionIdleTime, + String compactStrategy, + String buckets) + throws Exception { Map catalogOptions = catalog.options(); Map tableConf = StringUtils.isNullOrWhitespaceOnly(tableOptions) @@ -180,6 +205,10 @@ public String[] call( "You must specify 'order strategy' and 'order by columns' both."); } + if (buckets != null) { + action.withBucketsExpression(buckets); + } + if (!(StringUtils.isNullOrWhitespaceOnly(partitions))) { action.withPartitions(ParameterUtils.getPartitions(partitions.split(";"))); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java index 6a5d5ab14424..c6c97f43a319 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java @@ -47,7 +47,9 @@ import org.apache.paimon.table.PostponeUtils.CompactBucket; import org.apache.paimon.table.PostponeUtils.PostponeBucketNumResolver; import org.apache.paimon.table.sink.ChannelComputer; +import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.InternalRowPartitionComputer; +import org.apache.paimon.utils.ParameterUtils; import org.apache.paimon.utils.Pair; import org.apache.flink.api.common.RuntimeExecutionMode; @@ -65,6 +67,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; @@ -85,6 +88,9 @@ public class CompactAction extends TableActionBase { @Nullable protected Boolean fullCompaction; + private String bucketsExpression; + private Set buckets; + public CompactAction( String database, String tableName, @@ -127,6 +133,12 @@ public CompactAction withFullCompaction(Boolean fullCompaction) { return this; } + public CompactAction withBucketsExpression(String bucketsExpression) { + this.buckets = null; + this.bucketsExpression = bucketsExpression; + return this; + } + @Override public void build() throws Exception { buildImpl(); @@ -137,6 +149,7 @@ protected boolean buildImpl() throws Exception { boolean isStreaming = conf.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.STREAMING; FileStoreTable fileStoreTable = (FileStoreTable) table; + resolveBuckets(fileStoreTable); PartitionPredicate partitionPredicate = getPartitionPredicate(); if (fileStoreTable.coreOptions().bucket() == BucketMode.POSTPONE_BUCKET) { buildForPostponeBucketCompaction(env, fileStoreTable, isStreaming); @@ -206,6 +219,8 @@ protected void buildForBucketedTableCompact( .withBucketDistributionStrategy(bucketDistributionStrategy); sourceBuilder.withPartitionPredicate(getPartitionPredicate()); + sourceBuilder.withBucketFilter( + buckets == null ? null : new SpecifiedBucketFilter(buckets)); DataStreamSource source = sourceBuilder .withEnv(env) @@ -240,6 +255,25 @@ protected PartitionPredicate getPartitionPredicate() throws Exception { (FileStoreTable) table, partitions, whereSql, "compaction"); } + protected boolean bucketsSpecified() { + return bucketsExpression != null; + } + + private void resolveBuckets(FileStoreTable table) { + if (!bucketsSpecified()) { + buckets = null; + return; + } + checkArgument( + table.bucketMode() == BucketMode.HASH_FIXED, + "Specifying buckets is only supported for fixed-bucket tables, but the table bucket mode is %s.", + table.bucketMode()); + buckets = + new HashSet<>( + ParameterUtils.parseIntegerRanges( + bucketsExpression, table.coreOptions().bucket())); + } + protected boolean buildForPostponeBucketCompaction( StreamExecutionEnvironment env, FileStoreTable table, boolean isStreaming) { checkArgument( @@ -361,6 +395,23 @@ private boolean buildNothingToCompact(StreamExecutionEnvironment env) { return false; } + private static class SpecifiedBucketFilter + implements Filter, java.io.Serializable { + + private static final long serialVersionUID = 1L; + + private final Set buckets; + + private SpecifiedBucketFilter(Set buckets) { + this.buckets = buckets; + } + + @Override + public boolean test(Integer bucket) { + return buckets.contains(bucket); + } + } + private static class CompactBucketChannelComputer implements ChannelComputer { private static final long serialVersionUID = 1L; diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactActionFactory.java index dc9614ce344b..45057440282b 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactActionFactory.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactActionFactory.java @@ -37,6 +37,8 @@ public class CompactActionFactory implements ActionFactory { private static final String PARTITION_IDLE_TIME = "partition_idle_time"; + private static final String BUCKETS = "buckets"; + @Override public String identifier() { return IDENTIFIER; @@ -77,6 +79,10 @@ public Optional create(MultipleParameterToolAdapter params) { action.withWhereSql(params.get(WHERE)); } + if (params.has(BUCKETS)) { + action.withBucketsExpression(params.get(BUCKETS)); + } + return Optional.of(action); } @@ -107,7 +113,8 @@ public void printHelp() { + "[--table_conf =] \n" + "[--order_by ] \n" + "[--partition_idle_time ] \n" - + "[--compact_strategy ]"); + + "[--compact_strategy ] \n" + + "[--buckets ]"); System.out.println( " compact --warehouse s3://path/to/warehouse --database " + "--table [--catalog_conf [--catalog_conf ...]]"); @@ -135,6 +142,9 @@ public void printHelp() { System.out.println( " compact --warehouse hdfs:///path/to/warehouse --database test_db --table test_table " + "--partition_idle_time 10s"); + System.out.println( + " compact --warehouse hdfs:///path/to/warehouse --database test_db --table test_table " + + "--compact_strategy full --buckets 0-9,20"); System.out.println( "--compact_strategy determines how to pick files to be merged, the default is determined by the runtime execution mode. " + "`full` : Only supports batch mode. All files will be selected for merging." diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java index 289802e2ba6b..b491398b7063 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java @@ -67,6 +67,10 @@ public void run() throws Exception { @Override public void build() throws Exception { + if (bucketsSpecified()) { + throw new IllegalArgumentException( + "Specifying buckets is not supported for sort compact."); + } // only support batch sort yet if (env.getConfiguration().get(ExecutionOptions.RUNTIME_MODE) != RuntimeExecutionMode.BATCH) { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java index 74ae5f786d4e..902683002682 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java @@ -64,6 +64,10 @@ public class CompactProcedure extends ProcedureBase { @ArgumentHint( name = "compact_strategy", type = @DataTypeHint("STRING"), + isOptional = true), + @ArgumentHint( + name = "buckets", + type = @DataTypeHint("STRING"), isOptional = true) }) public String[] call( @@ -75,7 +79,8 @@ public String[] call( String tableOptions, String where, String partitionIdleTime, - String compactStrategy) + String compactStrategy, + String buckets) throws Exception { Map catalogOptions = catalog.options(); Map tableConf = @@ -119,6 +124,10 @@ public String[] call( "You must specify 'order strategy' and 'order by columns' both."); } + if (buckets != null) { + action.withBucketsExpression(buckets); + } + if (!(isNullOrWhitespaceOnly(partitions))) { action.withPartitions(getPartitions(partitions.split(";"))); } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java index 96961272f12a..9bc38ebbbb2e 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java @@ -33,6 +33,7 @@ import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.system.CompactBucketsTable; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.Preconditions; import org.apache.flink.api.common.eventtime.WatermarkStrategy; @@ -67,6 +68,7 @@ public class CompactorSourceBuilder { private boolean isContinuous = false; private StreamExecutionEnvironment env; @Nullable private PartitionPredicate partitionPredicate = null; + @Nullable private Filter bucketFilter = null; @Nullable private Duration partitionIdleTime = null; private CompactionBucketDistributionStrategy bucketDistributionStrategy = @@ -100,6 +102,9 @@ public CompactorSourceBuilder withPartitionIdleTime(@Nullable Duration partition if (partitionPredicate != null) { readBuilder.withPartitionFilter(partitionPredicate); } + if (bucketFilter != null) { + readBuilder.withBucketFilter(bucketFilter); + } if (CoreOptions.fromMap(table.options()).manifestDeleteFileDropStats()) { readBuilder = readBuilder.dropStats(); } @@ -231,6 +236,11 @@ public CompactorSourceBuilder withPartitionPredicate( return this; } + public CompactorSourceBuilder withBucketFilter(@Nullable Filter bucketFilter) { + this.bucketFilter = bucketFilter; + return this; + } + public CompactorSourceBuilder withBucketDistributionStrategy( CompactionBucketDistributionStrategy bucketDistributionStrategy) { this.bucketDistributionStrategy = bucketDistributionStrategy; diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java index fedd98ceea3b..a4fe02162dfe 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java @@ -23,6 +23,7 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.data.BinaryRow; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.flink.FlinkConnectorOptions; import org.apache.paimon.fs.Path; @@ -73,6 +74,7 @@ import java.util.UUID; import java.util.concurrent.ThreadLocalRandom; import java.util.stream.Collectors; +import java.util.stream.IntStream; import static org.apache.paimon.utils.CommonTestUtils.waitUtil; import static org.assertj.core.api.Assertions.assertThat; @@ -704,6 +706,68 @@ public void testTableConf() throws Exception { .isEqualTo("6"); } + @Test + public void testCompactSpecifiedBucketRangesFromAction() throws Exception { + Map tableOptions = new HashMap<>(); + tableOptions.put(CoreOptions.WRITE_ONLY.key(), "true"); + tableOptions.put(CoreOptions.BUCKET.key(), "4"); + FileStoreTable table = + prepareTable( + Collections.emptyList(), + Collections.singletonList("k"), + Collections.emptyList(), + tableOptions); + + writeData( + IntStream.range(0, 40) + .mapToObj( + i -> + rowData( + i, + 1, + 0, + BinaryString.fromString("first"))) + .toArray(GenericRow[]::new)); + writeData( + IntStream.range(40, 80) + .mapToObj( + i -> + rowData( + i, + 2, + 0, + BinaryString.fromString("second"))) + .toArray(GenericRow[]::new)); + + CompactAction action = + createAction( + CompactAction.class, + "compact", + "--warehouse", + warehouse, + "--database", + database, + "--table", + tableName, + "--compact_strategy", + "full", + "--buckets", + "0-1"); + StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build(); + action.withStreamExecutionEnvironment(env).build(); + env.execute(); + + Map filesPerBucket = + table.newSnapshotReader().read().dataSplits().stream() + .collect( + Collectors.toMap( + DataSplit::bucket, split -> split.dataFiles().size())); + assertThat(filesPerBucket.get(0)).isEqualTo(1); + assertThat(filesPerBucket.get(1)).isEqualTo(1); + assertThat(filesPerBucket.get(2)).isEqualTo(2); + assertThat(filesPerBucket.get(3)).isEqualTo(2); + } + @Test public void testSpecifyNonPartitionField() throws Exception { Map tableOptions = new HashMap<>(); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java index afe45cef66f5..ce163da4217c 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactProcedureITCase.java @@ -53,6 +53,46 @@ public class CompactProcedureITCase extends CatalogITCaseBase { // ----------------------- Non-sort Compact ----------------------- + @Test + public void testCompactSpecifiedBucketRanges() throws Exception { + sql( + "CREATE TABLE T (" + + " k INT," + + " v INT," + + " PRIMARY KEY (k) NOT ENFORCED" + + ") WITH (" + + " 'write-only' = 'true'," + + " 'bucket' = '4'" + + ")"); + FileStoreTable table = paimonTable("T"); + + sql( + "INSERT INTO T VALUES " + + IntStream.range(0, 40) + .mapToObj(i -> String.format("(%d, 1)", i)) + .collect(Collectors.joining(","))); + sql( + "INSERT INTO T VALUES " + + IntStream.range(40, 80) + .mapToObj(i -> String.format("(%d, 2)", i)) + .collect(Collectors.joining(","))); + + tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true); + sql( + "CALL sys.compact(`table` => 'default.T', compact_strategy => 'full', " + + "`buckets` => '0-1')"); + + Map filesPerBucket = + table.newSnapshotReader().read().dataSplits().stream() + .collect( + Collectors.toMap( + DataSplit::bucket, split -> split.dataFiles().size())); + assertThat(filesPerBucket.get(0)).isEqualTo(1); + assertThat(filesPerBucket.get(1)).isEqualTo(1); + assertThat(filesPerBucket.get(2)).isEqualTo(2); + assertThat(filesPerBucket.get(3)).isEqualTo(2); + } + @Test public void testBatchCompact() throws Exception { sql( diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java index b2817bce5cab..f10710fe3698 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java @@ -57,6 +57,7 @@ import org.apache.paimon.table.source.EndOfScanException; import org.apache.paimon.table.source.snapshot.SnapshotReader; import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.ParameterUtils; import org.apache.paimon.utils.ProcedureUtils; import org.apache.paimon.utils.SerializationUtils; import org.apache.paimon.utils.StringUtils; @@ -90,6 +91,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -110,7 +112,16 @@ * Compact procedure. Usage: * * 

- *  CALL sys.compact(table => 'tableId', [partitions => 'p1=0,p2=0;p1=0,p2=1'], [order_strategy => 'xxx'], [order_by => 'xxx'], [where => 'p1>0'])
+ *  CALL sys.compact(
+ *      table => 'tableId',
+ *      [partitions => 'p1=0,p2=0;p1=0,p2=1'],
+ *      [order_strategy => 'xxx'],
+ *      [order_by => 'xxx'],
+ *      [where => 'p1>0'],
+ *      [buckets => '0-99,200-299'])
+ *
+ *  -- Buckets support a single id, comma-separated ids, and closed ranges.
+ *  CALL sys.compact(table => 'tableId', compact_strategy => 'full', buckets => '0-99,200-299')
  *
*/ public class CompactProcedure extends BaseProcedure { @@ -127,6 +138,7 @@ public class CompactProcedure extends BaseProcedure { ProcedureParameter.optional("where", StringType), ProcedureParameter.optional("options", StringType), ProcedureParameter.optional("partition_idle_time", StringType), + ProcedureParameter.optional("buckets", StringType), }; private static final StructType OUTPUT_TYPE = @@ -166,6 +178,7 @@ public InternalRow[] call(InternalRow args) { String options = args.isNullAt(6) ? null : args.getString(6); Duration partitionIdleTime = blank(args, 7) ? null : TimeUtils.parseDuration(args.getString(7)); + String buckets = blank(args, 8) ? null : args.getString(8); if (OrderType.NONE.name().equals(sortType) && !sortColumns.isEmpty()) { throw new IllegalArgumentException( "order_strategy \"none\" cannot work with order_by columns."); @@ -233,7 +246,8 @@ public InternalRow[] call(InternalRow args) { sortColumns, relation, partitionPredicate, - partitionIdleTime)); + partitionIdleTime, + buckets)); return new InternalRow[] {internalRow}; }); } @@ -254,9 +268,26 @@ private boolean execute( List sortColumns, DataSourceV2Relation relation, @Nullable PartitionPredicate partitionPredicate, - @Nullable Duration partitionIdleTime) { + @Nullable Duration partitionIdleTime, + @Nullable String buckets) { BucketMode bucketMode = table.bucketMode(); OrderType orderType = OrderType.of(sortType); + final Set bucketSet; + if (buckets == null) { + bucketSet = null; + } else { + checkArgument( + bucketMode == BucketMode.HASH_FIXED, + "Specifying buckets is only supported for fixed-bucket tables, but the table bucket mode is %s.", + bucketMode); + checkArgument( + orderType == OrderType.NONE, + "Specifying buckets is not supported for sort compact."); + bucketSet = + new HashSet<>( + ParameterUtils.parseIntegerRanges( + buckets, table.coreOptions().bucket())); + } boolean clusterIncrementalEnabled = table.coreOptions().clusteringIncrementalEnabled(); if (compactStrategy == null) { @@ -288,6 +319,7 @@ private boolean execute( fullCompact, partitionPredicate, partitionIdleTime, + bucketSet, javaSparkContext); break; case BUCKET_UNAWARE: @@ -337,6 +369,7 @@ private void compactAwareBucketTable( boolean fullCompact, @Nullable PartitionPredicate partitionPredicate, @Nullable Duration partitionIdleTime, + @Nullable Set bucketSet, JavaSparkContext javaSparkContext) { SnapshotReader snapshotReader = table.newSnapshotReader(); if (partitionPredicate != null) { @@ -350,6 +383,7 @@ private void compactAwareBucketTable( snapshotReader.bucketEntries().stream() .map(entry -> Pair.of(entry.partition(), entry.bucket())) .distinct() + .filter(pair -> bucketSet == null || bucketSet.contains(pair.getRight())) .filter( pair -> !filterByPartitionIdleTime diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala index 77dbe8cb668c..32bf277ba7c0 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala @@ -477,6 +477,45 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT }) } + test("Paimon Procedure: compact specified bucket ranges") { + withTable("T") { + spark.sql( + """ + |CREATE TABLE T (id INT, value STRING) + |TBLPROPERTIES ('primary-key'='id', 'bucket'='4', 'write-only'='true') + |""".stripMargin) + + val table = loadTable("T") + spark.sql("INSERT INTO T SELECT id, 'first' FROM range(0, 40)") + spark.sql("INSERT INTO T SELECT id, 'second' FROM range(40, 80)") + + spark.sql( + "CALL sys.compact(table => 'T', compact_strategy => 'full', buckets => '0-1')") + + val filesPerBucket = table.newSnapshotReader.read.dataSplits.asScala + .map(split => split.bucket -> split.dataFiles.size) + .toMap + Assertions.assertThat(filesPerBucket(0)).isEqualTo(1) + Assertions.assertThat(filesPerBucket(1)).isEqualTo(1) + Assertions.assertThat(filesPerBucket(2)).isEqualTo(2) + Assertions.assertThat(filesPerBucket(3)).isEqualTo(2) + } + } + + test("Paimon Procedure: reject buckets for dynamic bucket table") { + withTable("T") { + spark.sql( + """ + |CREATE TABLE T (id INT, value STRING) + |TBLPROPERTIES ('primary-key'='id', 'bucket'='-1', 'write-only'='true') + |""".stripMargin) + + assertThatThrownBy( + () => spark.sql("CALL sys.compact(table => 'T', buckets => '0')").collect()) + .hasMessageContaining("Specifying buckets is only supported for fixed-bucket tables") + } + } + test("Paimon Procedure: compact aware bucket pk table with many small files") { Seq(3, -1).foreach( bucket => { From 582b98803150df0e07720879fb75945f30e2b383 Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Fri, 14 Aug 2026 17:08:23 +0800 Subject: [PATCH 2/3] fix code format --- .../apache/paimon/utils/ParameterUtils.java | 6 ++---- .../paimon/flink/action/CompactAction.java | 8 +++---- .../flink/procedure/CompactProcedure.java | 5 +---- .../flink/action/CompactActionITCase.java | 16 ++------------ .../procedure/CompactProcedureTestBase.scala | 21 ++++++++----------- 5 files changed, 17 insertions(+), 39 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java b/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java index d934a9ae6aa3..e740940ffea5 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java @@ -40,13 +40,11 @@ /** This is a util class for converting string parameter to another format. */ public class ParameterUtils { - private static final Pattern INTEGER_RANGE = - Pattern.compile("([0-9]+)(?:\\s*-\\s*([0-9]+))?"); + private static final Pattern INTEGER_RANGE = Pattern.compile("([0-9]+)(?:\\s*-\\s*([0-9]+))?"); public static List parseIntegerRanges(String values, int exclusiveUpperBound) { Preconditions.checkArgument( - !StringUtils.isNullOrWhitespaceOnly(values), - "Integer ranges must not be empty."); + !StringUtils.isNullOrWhitespaceOnly(values), "Integer ranges must not be empty."); Preconditions.checkArgument( exclusiveUpperBound > 0, "Exclusive upper bound must be greater than 0."); Set result = new LinkedHashSet<>(); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java index c6c97f43a319..1d01ba1f2e78 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java @@ -49,8 +49,8 @@ import org.apache.paimon.table.sink.ChannelComputer; import org.apache.paimon.utils.Filter; import org.apache.paimon.utils.InternalRowPartitionComputer; -import org.apache.paimon.utils.ParameterUtils; import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.ParameterUtils; import org.apache.flink.api.common.RuntimeExecutionMode; import org.apache.flink.configuration.ExecutionOptions; @@ -219,8 +219,7 @@ protected void buildForBucketedTableCompact( .withBucketDistributionStrategy(bucketDistributionStrategy); sourceBuilder.withPartitionPredicate(getPartitionPredicate()); - sourceBuilder.withBucketFilter( - buckets == null ? null : new SpecifiedBucketFilter(buckets)); + sourceBuilder.withBucketFilter(buckets == null ? null : new SpecifiedBucketFilter(buckets)); DataStreamSource source = sourceBuilder .withEnv(env) @@ -395,8 +394,7 @@ private boolean buildNothingToCompact(StreamExecutionEnvironment env) { return false; } - private static class SpecifiedBucketFilter - implements Filter, java.io.Serializable { + private static class SpecifiedBucketFilter implements Filter, java.io.Serializable { private static final long serialVersionUID = 1L; diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java index 902683002682..71c31338b7b4 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactProcedure.java @@ -65,10 +65,7 @@ public class CompactProcedure extends ProcedureBase { name = "compact_strategy", type = @DataTypeHint("STRING"), isOptional = true), - @ArgumentHint( - name = "buckets", - type = @DataTypeHint("STRING"), - isOptional = true) + @ArgumentHint(name = "buckets", type = @DataTypeHint("STRING"), isOptional = true) }) public String[] call( ProcedureContext procedureContext, diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java index a4fe02162dfe..d1bcf1db3a85 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java @@ -720,23 +720,11 @@ public void testCompactSpecifiedBucketRangesFromAction() throws Exception { writeData( IntStream.range(0, 40) - .mapToObj( - i -> - rowData( - i, - 1, - 0, - BinaryString.fromString("first"))) + .mapToObj(i -> rowData(i, 1, 0, BinaryString.fromString("first"))) .toArray(GenericRow[]::new)); writeData( IntStream.range(40, 80) - .mapToObj( - i -> - rowData( - i, - 2, - 0, - BinaryString.fromString("second"))) + .mapToObj(i -> rowData(i, 2, 0, BinaryString.fromString("second"))) .toArray(GenericRow[]::new)); CompactAction action = diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala index 32bf277ba7c0..b4facd1a4ea3 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala @@ -479,18 +479,16 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT test("Paimon Procedure: compact specified bucket ranges") { withTable("T") { - spark.sql( - """ - |CREATE TABLE T (id INT, value STRING) - |TBLPROPERTIES ('primary-key'='id', 'bucket'='4', 'write-only'='true') - |""".stripMargin) + spark.sql(""" + |CREATE TABLE T (id INT, value STRING) + |TBLPROPERTIES ('primary-key'='id', 'bucket'='4', 'write-only'='true') + |""".stripMargin) val table = loadTable("T") spark.sql("INSERT INTO T SELECT id, 'first' FROM range(0, 40)") spark.sql("INSERT INTO T SELECT id, 'second' FROM range(40, 80)") - spark.sql( - "CALL sys.compact(table => 'T', compact_strategy => 'full', buckets => '0-1')") + spark.sql("CALL sys.compact(table => 'T', compact_strategy => 'full', buckets => '0-1')") val filesPerBucket = table.newSnapshotReader.read.dataSplits.asScala .map(split => split.bucket -> split.dataFiles.size) @@ -504,11 +502,10 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT test("Paimon Procedure: reject buckets for dynamic bucket table") { withTable("T") { - spark.sql( - """ - |CREATE TABLE T (id INT, value STRING) - |TBLPROPERTIES ('primary-key'='id', 'bucket'='-1', 'write-only'='true') - |""".stripMargin) + spark.sql(""" + |CREATE TABLE T (id INT, value STRING) + |TBLPROPERTIES ('primary-key'='id', 'bucket'='-1', 'write-only'='true') + |""".stripMargin) assertThatThrownBy( () => spark.sql("CALL sys.compact(table => 'T', buckets => '0')").collect()) From f5e0aff8164ef634ff763572a96e716d6eec16bf Mon Sep 17 00:00:00 2001 From: sanshi <1715734693@qq.com> Date: Fri, 14 Aug 2026 20:50:44 +0800 Subject: [PATCH 3/3] fix compactDatabase procedure --- .../paimon/spark/procedure/CompactDatabaseProcedure.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactDatabaseProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactDatabaseProcedure.java index 44889a8cb5a4..a68eee65dd24 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactDatabaseProcedure.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactDatabaseProcedure.java @@ -181,7 +181,7 @@ private void compactTable(String tableName, String options) throws Exception { // Create InternalRow with the parameters for CompactProcedure // Parameters: table, partitions, compact_strategy, order_strategy, order_by, where, - // options, partition_idle_time + // options, partition_idle_time, buckets InternalRow compactArgs = newInternalRow( UTF8String.fromString(tableName), // table @@ -191,7 +191,8 @@ private void compactTable(String tableName, String options) throws Exception { null, // order_by null, // where options == null ? null : UTF8String.fromString(options), // options - null // partition_idle_time + null, // partition_idle_time + null // buckets ); InternalRow[] result = compactProcedure.call(compactArgs);