[core] Support row-count based bucket calculation for postpone compact.

docs/docs/primary-key-table/data-distribution.md

@@ -79,6 +79,13 @@ you need to run a compaction job.
 See `compact` [procedure](../flink/procedures).
 The bucket number for the partitions compacted for the first time
 is configured by the option `postpone.default-bucket-num`, whose default value is `1`.
+You can also configure `postpone.target-row-num-per-bucket` to calculate the bucket number
+from the row count of the files in the postpone bucket directory.
+The calculated bucket number is `ceil(row_count / postpone.target-row-num-per-bucket)`,
+and is at least `1`.
+When this option is configured, it takes precedence over `postpone.default-bucket-num`
+for partitions compacted for the first time.
+Partitions that already have real bucket files keep their existing bucket number.
 
 Finally, when you feel that the bucket number of some partition is too small,
 you can also run a rescale job.

docs/generated/core_configuration.html

@@ -1205,6 +1205,12 @@
             <td>Integer</td>
             <td>Bucket number for the partitions compacted for the first time in postpone bucket tables.</td>
         </tr>
+        <tr>
+            <td><h5>postpone.target-row-num-per-bucket</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Long</td>
+            <td>Target row number per bucket for partitions compacted from postpone bucket files for the first time.</td>
+        </tr>
         <tr>
             <td><h5>primary-key</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2557,6 +2557,13 @@ public String toString() {
                     .withDescription(
                             "Bucket number for the partitions compacted for the first time in postpone bucket tables.");
 
+    public static final ConfigOption<Long> POSTPONE_TARGET_ROW_NUM_PER_BUCKET =
+            key("postpone.target-row-num-per-bucket")
+                    .longType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Target row number per bucket for partitions compacted from postpone bucket files for the first time.");
+
     public static final ConfigOption<Long> GLOBAL_INDEX_ROW_COUNT_PER_SHARD =
             key("global-index.row-count-per-shard")
                     .longType()
@@ -4109,6 +4116,10 @@ public int postponeDefaultBucketNum() {
         return options.get(POSTPONE_DEFAULT_BUCKET_NUM);
     }
 
+    public Optional<Long> postponeTargetRowNumPerBucket() {
+        return options.getOptional(POSTPONE_TARGET_ROW_NUM_PER_BUCKET);
+    }
+
     public long globalIndexRowCountPerShard() {
         return options.get(GLOBAL_INDEX_ROW_COUNT_PER_SHARD);
     }

paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java

@@ -19,19 +19,74 @@
 package org.apache.paimon.table;
 
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.manifest.SimpleFileEntry;
 
+import javax.annotation.Nullable;
+
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.apache.paimon.CoreOptions.BUCKET;
 import static org.apache.paimon.CoreOptions.WRITE_ONLY;
 
 /** Utils for postpone table. */
 public class PostponeUtils {
 
+    public static int computeBucketNumByRowCount(long rowCount, long targetRowNumPerBucket) {
+        if (targetRowNumPerBucket <= 0) {
+            throw new IllegalArgumentException(
+                    "Option 'postpone.target-row-num-per-bucket' must be greater than 0.");
+        }
+
+        long bucketNum = rowCount <= 0 ? 1 : (rowCount - 1) / targetRowNumPerBucket + 1;
+        if (bucketNum > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException(
+                    "Computed postpone bucket number "
+                            + bucketNum
+                            + " exceeds the maximum integer value (Integer.MAX_VALUE = "
+                            + Integer.MAX_VALUE
+                            + "). Consider increasing 'postpone.target-row-num-per-bucket' "
+                            + "to reduce the bucket count.");
+        }
+        return (int) bucketNum;
+    }
+
+    public static int determineBucketNum(
+            BinaryRow partition,
+            Map<BinaryRow, Integer> knownNumBuckets,
+            Optional<Long> targetRowNumPerBucket,
+            Map<BinaryRow, Long> postponeRowCounts,
+            int defaultBucketNum) {
+        return determineBucketNum(
+                partition,
+                knownNumBuckets,
+                targetRowNumPerBucket.orElse(null),
+                postponeRowCounts,
+                defaultBucketNum);
+    }
+
+    public static int determineBucketNum(
+            BinaryRow partition,
+            Map<BinaryRow, Integer> knownNumBuckets,
+            @Nullable Long targetRowNumPerBucket,
+            Map<BinaryRow, Long> postponeRowCounts,
+            int defaultBucketNum) {
+        Integer knownBucketNum = knownNumBuckets.get(partition);
+        if (knownBucketNum != null) {
+            return knownBucketNum;
+        } else if (targetRowNumPerBucket != null) {
+            return computeBucketNumByRowCount(
+                    postponeRowCounts.getOrDefault(partition, 0L), targetRowNumPerBucket);
+        } else {
+            return defaultBucketNum;
+        }
+    }
+
     public static Map<BinaryRow, Integer> getKnownNumBuckets(FileStoreTable table) {
         Map<BinaryRow, Integer> knownNumBuckets = new HashMap<>();
         List<SimpleFileEntry> simpleFileEntries =
@@ -54,6 +109,18 @@ public static Map<BinaryRow, Integer> getKnownNumBuckets(FileStoreTable table) {
         return knownNumBuckets;
     }
 
+    /** Returns row counts of current active files in the postpone bucket. */
+    public static Map<BinaryRow, Long> getPostponeRowCounts(FileStoreTable table) {
+        Map<BinaryRow, Long> rowCounts = new HashMap<>();
+        Iterator<ManifestEntry> iterator =
+                table.newSnapshotReader().withBucket(BucketMode.POSTPONE_BUCKET).readFileIterator();
+        while (iterator.hasNext()) {
+            ManifestEntry entry = iterator.next();
+            rowCounts.merge(entry.partition(), entry.file().rowCount(), Long::sum);
+        }
+        return rowCounts;
+    }
+
     public static FileStoreTable tableForFixBucketWrite(FileStoreTable table) {
         Map<String, String> batchWriteOptions = new HashMap<>();
         batchWriteOptions.put(WRITE_ONLY.key(), "true");

paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java

@@ -0,0 +1,99 @@
+/*
+ * 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.table;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link PostponeUtils}. */
+public class PostponeUtilsTest {
+
+    @Test
+    public void testComputeBucketNumByRowCount() {
+        assertThat(PostponeUtils.computeBucketNumByRowCount(0, 100)).isEqualTo(1);
+        assertThat(PostponeUtils.computeBucketNumByRowCount(1, 100)).isEqualTo(1);
+        assertThat(PostponeUtils.computeBucketNumByRowCount(100, 100)).isEqualTo(1);
+        assertThat(PostponeUtils.computeBucketNumByRowCount(101, 100)).isEqualTo(2);
+        assertThat(PostponeUtils.computeBucketNumByRowCount(999, 200)).isEqualTo(5);
+        assertThat(PostponeUtils.computeBucketNumByRowCount(1000, 200)).isEqualTo(5);
+    }
+
+    @Test
+    public void testComputeBucketNumByRowCountRejectsInvalidTarget() {
+        assertThatThrownBy(() -> PostponeUtils.computeBucketNumByRowCount(100, 0))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining(
+                        "Option 'postpone.target-row-num-per-bucket' must be greater than 0.");
+    }
+
+    @Test
+    public void testComputeBucketNumByRowCountRejectsOverflow() {
+        assertThatThrownBy(() -> PostponeUtils.computeBucketNumByRowCount(Long.MAX_VALUE, 1))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("exceeds the maximum integer value")
+                .hasMessageContaining("Consider increasing 'postpone.target-row-num-per-bucket'");
+    }
+
+    @Test
+    public void testDetermineBucketNum() {
+        Map<BinaryRow, Integer> knownNumBuckets = new HashMap<>();
+        Map<BinaryRow, Long> postponeRowCounts = new HashMap<>();
+
+        BinaryRow knownPartition = partition(1);
+        BinaryRow targetPartition = partition(2);
+        BinaryRow defaultPartition = partition(3);
+
+        knownNumBuckets.put(knownPartition, 4);
+        postponeRowCounts.put(knownPartition, 1000L);
+        postponeRowCounts.put(targetPartition, 450L);
+
+        assertThat(
+                        PostponeUtils.determineBucketNum(
+                                knownPartition, knownNumBuckets, 200L, postponeRowCounts, 1))
+                .isEqualTo(4);
+        assertThat(
+                        PostponeUtils.determineBucketNum(
+                                targetPartition, knownNumBuckets, 200L, postponeRowCounts, 1))
+                .isEqualTo(3);
+        assertThat(
+                        PostponeUtils.determineBucketNum(
+                                defaultPartition,
+                                knownNumBuckets,
+                                (Long) null,
+                                postponeRowCounts,
+                                7))
+                .isEqualTo(7);
+    }
+
+    private static BinaryRow partition(int value) {
+        BinaryRow row = new BinaryRow(1);
+        BinaryRowWriter writer = new BinaryRowWriter(row);
+        writer.writeInt(0, value);
+        writer.complete();
+        return row;
+    }
+}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java

@@ -36,7 +36,6 @@
 import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
 import org.apache.paimon.flink.sink.RowDataChannelComputer;
 import org.apache.paimon.flink.source.CompactorSourceBuilder;
-import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.PartitionPredicate;
 import org.apache.paimon.predicate.PartitionPredicateVisitor;
@@ -45,6 +44,7 @@
 import org.apache.paimon.predicate.PredicateProjectionConverter;
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.PostponeUtils;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.InternalRowPartitionComputer;
 import org.apache.paimon.utils.Pair;
@@ -66,10 +66,10 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.apache.paimon.partition.PartitionPredicate.createBinaryPartitions;
 import static org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
@@ -291,6 +291,13 @@ protected boolean buildForPostponeBucketCompaction(
 
         Options options = new Options(table.options());
         int defaultBucketNum = options.get(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM);
+        Optional<Long> targetRowNumPerBucket =
+                options.getOptional(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET);
+        Map<BinaryRow, Integer> knownNumBuckets = PostponeUtils.getKnownNumBuckets(table);
+        Map<BinaryRow, Long> postponeRowCounts =
+                targetRowNumPerBucket.isPresent()
+                        ? PostponeUtils.getPostponeRowCounts(table)
+                        : Collections.emptyMap();
 
         // change bucket to a positive value, so we can scan files from the bucket = -2 directory
         Map<String, String> bucketOptions = new HashMap<>(table.options());
@@ -322,16 +329,13 @@ protected boolean buildForPostponeBucketCompaction(
         String commitUser = CoreOptions.createCommitUser(options);
         List<DataStream<Committable>> dataStreams = new ArrayList<>();
         for (BinaryRow partition : partitions) {
-            int bucketNum = defaultBucketNum;
-
-            Iterator<ManifestEntry> it =
-                    table.newSnapshotReader()
-                            .withPartitionFilter(Collections.singletonList(partition))
-                            .onlyReadRealBuckets()
-                            .readFileIterator();
-            if (it.hasNext()) {
-                bucketNum = it.next().totalBuckets();
-            }
+            int bucketNum =
+                    PostponeUtils.determineBucketNum(
+                            partition,
+                            knownNumBuckets,
+                            targetRowNumPerBucket,
+                            postponeRowCounts,
+                            defaultBucketNum);
 
             bucketOptions = new HashMap<>(table.options());
             bucketOptions.put(CoreOptions.BUCKET.key(), String.valueOf(bucketNum));

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java

@@ -438,6 +438,57 @@ public void testRescaleBucket() throws Exception {
         assertThat(collect(tEnv.executeSql(query))).hasSameElementsAs(expectedData);
     }
 
+    @Test
+    public void testCompactWithTargetRowNumPerBucket() throws Exception {
+        String warehouse = getTempDirPath();
+        TableEnvironment tEnv =
+                tableEnvironmentBuilder()
+                        .batchMode()
+                        .setConf(TableConfigOptions.TABLE_DML_SYNC, true)
+                        .build();
+
+        tEnv.executeSql(
+                "CREATE CATALOG mycat WITH (\n"
+                        + "  'type' = 'paimon',\n"
+                        + "  'warehouse' = '"
+                        + warehouse
+                        + "'\n"
+                        + ")");
+        tEnv.executeSql("USE CATALOG mycat");
+        tEnv.executeSql(
+                "CREATE TABLE T (\n"
+                        + "  pt INT,\n"
+                        + "  k INT,\n"
+                        + "  v INT,\n"
+                        + "  PRIMARY KEY (pt, k) NOT ENFORCED\n"
+                        + ") PARTITIONED BY (pt) WITH (\n"
+                        + "  'bucket' = '-2',\n"
+                        + "  'postpone.target-row-num-per-bucket' = '200',\n"
+                        + "  'postpone.batch-write-fixed-bucket' = 'false'\n"
+                        + ")");
+
+        List<String> values = new ArrayList<>();
+        for (int j = 0; j < 100; j++) {
+            values.add(String.format("(0, %d, %d)", j, j));
+        }
+        for (int j = 0; j < 450; j++) {
+            values.add(String.format("(1, %d, %d)", j, j));
+        }
+        tEnv.executeSql("INSERT INTO T VALUES " + String.join(", ", values)).await();
+        assertThat(collect(tEnv.executeSql("SELECT * FROM T"))).isEmpty();
+
+        tEnv.executeSql("CALL sys.compact(`table` => 'default.T')").await();
+
+        assertThat(collect(tEnv.executeSql("SELECT pt, COUNT(*) FROM T GROUP BY pt")))
+                .containsExactlyInAnyOrder("+I[0, 100]", "+I[1, 450]");
+        assertThat(
+                        collect(
+                                tEnv.executeSql(
+                                        "SELECT `partition`, COUNT(DISTINCT bucket) FROM `T$files` "
+                                                + "GROUP BY `partition`")))
+                .containsExactlyInAnyOrder("+I[{0}, 1]", "+I[{1}, 3]");
+    }
+
     @Timeout(TIMEOUT)
     @Test
     public void testInputChangelogProducer() throws Exception {