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 b9d0a461d4 [core] Support row-count based bucket calculation for
postpone compact. (#8434)
b9d0a461d4 is described below
commit b9d0a461d47af17f544bda70db79fb3b92e01a85
Author: Wenchao Wu <[email protected]>
AuthorDate: Tue Jul 7 15:58:10 2026 +0800
[core] Support row-count based bucket calculation for postpone compact.
(#8434)
Support calculating postpone bucket count by configured target row count
per bucket.
A new table option is added:
- `postpone.target-row-num-per-bucket`
When compacting postpone bucket files, bucket count is resolved in this
order:
1. Reuse known bucket count if the partition was already compacted
before.
2. If `postpone.target-row-num-per-bucket` is configured, compute bucket
count from active postpone row count.
3. Fall back to `postpone.default-bucket-num`.
---
docs/docs/primary-key-table/data-distribution.md | 7 ++
docs/generated/core_configuration.html | 6 ++
.../main/java/org/apache/paimon/CoreOptions.java | 11 +++
.../org/apache/paimon/table/PostponeUtils.java | 67 +++++++++++++++
.../org/apache/paimon/table/PostponeUtilsTest.java | 99 ++++++++++++++++++++++
.../apache/paimon/flink/action/CompactAction.java | 28 +++---
.../paimon/flink/PostponeBucketTableITCase.java | 51 +++++++++++
.../procedure/SparkPostponeCompactProcedure.scala | 37 +++++++-
.../paimon/spark/sql/PostponeBucketTableTest.scala | 41 +++++++++
9 files changed, 334 insertions(+), 13 deletions(-)
diff --git a/docs/docs/primary-key-table/data-distribution.md
b/docs/docs/primary-key-table/data-distribution.md
index 95e8c01aa6..49846b2ea3 100644
--- a/docs/docs/primary-key-table/data-distribution.md
+++ b/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.
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 2bb0099a49..6fa06a5ce3 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1217,6 +1217,12 @@ This config option does not affect the default
filesystem metastore.</td>
<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>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 0302d10995..85ed36dace 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2593,6 +2593,13 @@ public class CoreOptions implements Serializable {
.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()
@@ -4160,6 +4167,10 @@ public class CoreOptions implements Serializable {
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);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
index 097921df54..62fb41a9e9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java
@@ -19,12 +19,17 @@
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;
@@ -32,6 +37,56 @@ 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 class PostponeUtils {
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");
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java
new file mode 100644
index 0000000000..50d9bc2077
--- /dev/null
+++ b/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;
+ }
+}
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 7c46763975..cbdcf825f2 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
@@ -36,7 +36,6 @@ import org.apache.paimon.flink.sink.FlinkSinkBuilder;
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.PredicateBuilder;
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.time.Duration;
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 @@ public class CompactAction extends TableActionBase {
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 @@ public class CompactAction extends TableActionBase {
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));
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
index d285fbb044..d9f8781da3 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java
@@ -438,6 +438,57 @@ public class PostponeBucketTableITCase extends
AbstractTestBase {
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 {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
index d714f47541..4725ce09f9 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/procedure/SparkPostponeCompactProcedure.scala
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BinaryRow
import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.postpone.BucketFiles
+import org.apache.paimon.spark.PaimonImplicits._
import org.apache.paimon.spark.commands.{EncoderSerDeGroup,
PostponeFixBucketProcessor}
import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL,
ROW_KIND_COL}
import org.apache.paimon.spark.util.{ScanPlanHelper, SparkRowUtils}
@@ -63,13 +64,26 @@ case class SparkPostponeCompactProcedure(
// Create bucket computer to determine bucket count for each partition
private lazy val postponePartitionBucketComputer = {
val knownNumBuckets = PostponeUtils.getKnownNumBuckets(table)
+ val targetRowNumPerBucket: Option[java.lang.Long] =
+ table.coreOptions.postponeTargetRowNumPerBucket
+ val postponeRowCounts =
+ if (targetRowNumPerBucket.isDefined) {
+ PostponeUtils.getPostponeRowCounts(table)
+ } else {
+ Collections.emptyMap[BinaryRow, java.lang.Long]()
+ }
val defaultBucketNum =
if
(table.coreOptions.toConfiguration.contains(CoreOptions.POSTPONE_DEFAULT_BUCKET_NUM))
{
table.coreOptions.postponeDefaultBucketNum
} else {
spark.sparkContext.defaultParallelism
}
- (p: BinaryRow) => knownNumBuckets.getOrDefault(p, defaultBucketNum)
+
+ SparkPostponeCompactProcedure.PostponePartitionBucketComputer(
+ knownNumBuckets,
+ targetRowNumPerBucket,
+ postponeRowCounts,
+ defaultBucketNum)
}
private def partitionCols(df: DataFrame): Seq[Column] = {
@@ -237,3 +251,24 @@ case class SparkPostponeCompactProcedure(
LOG.info("Successfully committed postpone bucket compaction for table:
{}.", table.name())
}
}
+
+object SparkPostponeCompactProcedure {
+
+ private[procedure] case class PostponePartitionBucketComputer(
+ knownNumBuckets: java.util.Map[BinaryRow, Integer],
+ targetRowNumPerBucket: Option[java.lang.Long],
+ postponeRowCounts: java.util.Map[BinaryRow, java.lang.Long],
+ defaultBucketNum: Int)
+ extends (BinaryRow => Integer)
+ with Serializable {
+
+ override def apply(p: BinaryRow): Integer = {
+ PostponeUtils.determineBucketNum(
+ p,
+ knownNumBuckets,
+ targetRowNumPerBucket.orNull,
+ postponeRowCounts,
+ defaultBucketNum)
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
index d984f1385c..1bc3cc1888 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PostponeBucketTableTest.scala
@@ -292,6 +292,47 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
}
}
+ test("Postpone partition bucket table: compact with target row num per
bucket") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.target-row-num-per-bucket' = '200',
+ | 'postpone.batch-write-fixed-bucket' = 'false'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(4) */
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |CASE WHEN id < 100 THEN 0 ELSE 1 END AS pt
+ |FROM range (0, 550)
+ |""".stripMargin)
+
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(0)))
+ checkAnswer(sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY
bucket"), Seq(Row(-2)))
+
+ sql("CALL sys.compact(table => 't')")
+
+ checkAnswer(sql("SELECT count(*) FROM t"), Seq(Row(550)))
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{0}'
ORDER BY bucket"),
+ Seq(Row(0))
+ )
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{1}'
ORDER BY bucket"),
+ Seq(Row(0), Row(1), Row(2))
+ )
+ }
+ }
+
test("Postpone bucket table: skip clustering in writing phase") {
withTable("t") {
sql("""
