This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8db20324e9 [spark][flink] Support bucket-level compaction for
fixed-bucket tables (#9219)
8db20324e9 is described below
commit 8db20324e9cb11183484a07f3109aaf40963ffe0
Author: sanshi <[email protected]>
AuthorDate: Sat Aug 15 20:17:49 2026 +0800
[spark][flink] Support bucket-level compaction for fixed-bucket tables
(#9219)
---
.../org/apache/paimon/utils/ParameterUtils.java | 38 +++++++++++++++
.../apache/paimon/utils/ParameterUtilsTest.java | 55 ++++++++++++++++++++++
.../paimon/flink/procedure/CompactProcedure.java | 29 ++++++++++++
.../apache/paimon/flink/action/CompactAction.java | 49 +++++++++++++++++++
.../paimon/flink/action/CompactActionFactory.java | 12 ++++-
.../paimon/flink/action/SortCompactAction.java | 4 ++
.../paimon/flink/procedure/CompactProcedure.java | 10 +++-
.../flink/source/CompactorSourceBuilder.java | 10 ++++
.../paimon/flink/action/CompactActionITCase.java | 52 ++++++++++++++++++++
.../flink/procedure/CompactProcedureITCase.java | 40 ++++++++++++++++
.../spark/procedure/CompactDatabaseProcedure.java | 5 +-
.../paimon/spark/procedure/CompactProcedure.java | 40 ++++++++++++++--
.../spark/procedure/CompactProcedureTestBase.scala | 36 ++++++++++++++
13 files changed, 372 insertions(+), 8 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 2eaa4ba94c..e740940ffe 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,50 @@ import javax.annotation.Nullable;
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<Integer> 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<Integer> 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<Map<String, String>> getPartitions(String...
partitionStrings) {
List<Map<String, String>> 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 0000000000..47f1be885f
--- /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 18e03e053c..52e4d4ba2c 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 class CompactProcedure extends ProcedureBase {
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<String, String> catalogOptions = catalog.options();
Map<String, String> tableConf =
StringUtils.isNullOrWhitespaceOnly(tableOptions)
@@ -180,6 +205,10 @@ public class CompactProcedure extends ProcedureBase {
"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 6a5d5ab144..1d01ba1f2e 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,8 +47,10 @@ import org.apache.paimon.table.PostponeUtils;
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.Pair;
+import org.apache.paimon.utils.ParameterUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ExecutionOptions;
@@ -65,6 +67,7 @@ import java.time.Duration;
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<Integer> buckets;
+
public CompactAction(
String database,
String tableName,
@@ -127,6 +133,12 @@ public class CompactAction extends TableActionBase {
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 @@ public class CompactAction extends TableActionBase {
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,7 @@ public class CompactAction extends TableActionBase {
.withBucketDistributionStrategy(bucketDistributionStrategy);
sourceBuilder.withPartitionPredicate(getPartitionPredicate());
+ sourceBuilder.withBucketFilter(buckets == null ? null : new
SpecifiedBucketFilter(buckets));
DataStreamSource<RowData> source =
sourceBuilder
.withEnv(env)
@@ -240,6 +254,25 @@ public class CompactAction extends TableActionBase {
(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 +394,22 @@ public class CompactAction extends TableActionBase {
return false;
}
+ private static class SpecifiedBucketFilter implements Filter<Integer>,
java.io.Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Set<Integer> buckets;
+
+ private SpecifiedBucketFilter(Set<Integer> buckets) {
+ this.buckets = buckets;
+ }
+
+ @Override
+ public boolean test(Integer bucket) {
+ return buckets.contains(bucket);
+ }
+ }
+
private static class CompactBucketChannelComputer implements
ChannelComputer<CompactBucket> {
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 dc9614ce34..4505744028 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 class CompactActionFactory implements ActionFactory {
action.withWhereSql(params.get(WHERE));
}
+ if (params.has(BUCKETS)) {
+ action.withBucketsExpression(params.get(BUCKETS));
+ }
+
return Optional.of(action);
}
@@ -107,7 +113,8 @@ public class CompactActionFactory implements ActionFactory {
+ "[--table_conf <key>=<value>] \n"
+ "[--order_by <order_columns>] \n"
+ "[--partition_idle_time <partition_idle_time>] \n"
- + "[--compact_strategy <compact_strategy>]");
+ + "[--compact_strategy <compact_strategy>] \n"
+ + "[--buckets <bucket_ids_or_ranges>]");
System.out.println(
" compact --warehouse s3://path/to/warehouse --database
<database_name> "
+ "--table <table_name> [--catalog_conf
<paimon_catalog_conf> [--catalog_conf <paimon_catalog_conf> ...]]");
@@ -135,6 +142,9 @@ public class CompactActionFactory implements ActionFactory {
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 289802e2ba..b491398b70 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 class SortCompactAction extends CompactAction {
@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 74ae5f786d..71c31338b7 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,7 +64,8 @@ public class CompactProcedure extends ProcedureBase {
@ArgumentHint(
name = "compact_strategy",
type = @DataTypeHint("STRING"),
- isOptional = true)
+ isOptional = true),
+ @ArgumentHint(name = "buckets", type =
@DataTypeHint("STRING"), isOptional = true)
})
public String[] call(
ProcedureContext procedureContext,
@@ -75,7 +76,8 @@ public class CompactProcedure extends ProcedureBase {
String tableOptions,
String where,
String partitionIdleTime,
- String compactStrategy)
+ String compactStrategy,
+ String buckets)
throws Exception {
Map<String, String> catalogOptions = catalog.options();
Map<String, String> tableConf =
@@ -119,6 +121,10 @@ public class CompactProcedure extends ProcedureBase {
"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 96961272f1..9bc38ebbbb 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.DataSplit;
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<Integer> bucketFilter = null;
@Nullable private Duration partitionIdleTime = null;
private CompactionBucketDistributionStrategy bucketDistributionStrategy =
@@ -100,6 +102,9 @@ public class CompactorSourceBuilder {
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 class CompactorSourceBuilder {
return this;
}
+ public CompactorSourceBuilder withBucketFilter(@Nullable Filter<Integer>
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 fedd98ceea..d1bcf1db3a 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.CoreOptions;
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.Map;
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,56 @@ public class CompactActionITCase extends
CompactActionITCaseBase {
.isEqualTo("6");
}
+ @Test
+ public void testCompactSpecifiedBucketRangesFromAction() throws Exception {
+ Map<String, String> 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<Integer, Integer> 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<String, String> 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 afe45cef66..ce163da421 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 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
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<Integer, Integer> 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/CompactDatabaseProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactDatabaseProcedure.java
index 44889a8cb5..a68eee65dd 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 @@ public class CompactDatabaseProcedure extends BaseProcedure
{
// 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 @@ public class CompactDatabaseProcedure extends BaseProcedure
{
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);
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 f9368cb50a..832bd54780 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
@@ -58,6 +58,7 @@ import org.apache.paimon.table.source.DataSplit;
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;
@@ -91,6 +92,7 @@ import java.util.ArrayList;
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;
@@ -112,7 +114,16 @@ import static
org.apache.spark.sql.types.DataTypes.StringType;
* Compact procedure. Usage:
*
* <pre><code>
- * 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')
* </code></pre>
*/
public class CompactProcedure extends BaseProcedure {
@@ -129,6 +140,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 =
@@ -168,6 +180,7 @@ public class CompactProcedure extends BaseProcedure {
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.");
@@ -235,7 +248,8 @@ public class CompactProcedure extends BaseProcedure {
sortColumns,
relation,
partitionPredicate,
- partitionIdleTime));
+ partitionIdleTime,
+ buckets));
return new InternalRow[] {internalRow};
});
}
@@ -256,9 +270,26 @@ public class CompactProcedure extends BaseProcedure {
List<String> 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<Integer> 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) {
@@ -290,6 +321,7 @@ public class CompactProcedure extends BaseProcedure {
fullCompact,
partitionPredicate,
partitionIdleTime,
+ bucketSet,
javaSparkContext);
break;
case BUCKET_UNAWARE:
@@ -343,6 +375,7 @@ public class CompactProcedure extends BaseProcedure {
boolean fullCompact,
@Nullable PartitionPredicate partitionPredicate,
@Nullable Duration partitionIdleTime,
+ @Nullable Set<Integer> bucketSet,
JavaSparkContext javaSparkContext) {
SnapshotReader snapshotReader = table.newSnapshotReader();
if (partitionPredicate != null) {
@@ -356,6 +389,7 @@ public class CompactProcedure extends BaseProcedure {
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 9c5f464bbc..a361fcae44 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
@@ -489,6 +489,42 @@ 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 => {
