[spark][flink] Support bucket-level compaction for fixed-bucket tables

paimon-common/src/main/java/org/apache/paimon/utils/ParameterUtils.java

@@ -30,12 +30,50 @@
 
 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) {

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");
+    }
+}

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<String, String> catalogOptions = catalog.options();
         Map<String, String> 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(";")));
         }

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.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.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 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,7 @@ protected void buildForBucketedTableCompact(
                         .withBucketDistributionStrategy(bucketDistributionStrategy);
 
         sourceBuilder.withPartitionPredicate(getPartitionPredicate());
+        sourceBuilder.withBucketFilter(buckets == null ? null : new SpecifiedBucketFilter(buckets));
         DataStreamSource<RowData> source =
                 sourceBuilder
                         .withEnv(env)
@@ -240,6 +254,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 +394,22 @@ private boolean buildNothingToCompact(StreamExecutionEnvironment env) {
         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;

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<Action> 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 <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 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."

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) {

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 String[] call(
             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 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(";")));
         }

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<Integer> 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<Integer> bucketFilter) {
+        this.bucketFilter = bucketFilter;
+        return this;
+    }
+
     public CompactorSourceBuilder withBucketDistributionStrategy(
             CompactionBucketDistributionStrategy bucketDistributionStrategy) {
         this.bucketDistributionStrategy = bucketDistributionStrategy;