diff --git a/docs/docs/primary-key-table/data-distribution.md b/docs/docs/primary-key-table/data-distribution.md index 95e8c01aa6c9..49846b2ea3b8 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 33d038ba43de..15b4b2492177 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1205,6 +1205,12 @@ Integer Bucket number for the partitions compacted for the first time in postpone bucket tables. + +
postpone.target-row-num-per-bucket
+ (none) + Long + Target row number per bucket for partitions compacted from postpone bucket files for the first time. +
primary-key
(none) 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 55c25b409b71..62cc42eefd3a 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/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 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 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 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 097921df54e2..62fb41a9e94b 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 @@ /** 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 knownNumBuckets, + Optional targetRowNumPerBucket, + Map postponeRowCounts, + int defaultBucketNum) { + return determineBucketNum( + partition, + knownNumBuckets, + targetRowNumPerBucket.orElse(null), + postponeRowCounts, + defaultBucketNum); + } + + public static int determineBucketNum( + BinaryRow partition, + Map knownNumBuckets, + @Nullable Long targetRowNumPerBucket, + Map 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 getKnownNumBuckets(FileStoreTable table) { Map knownNumBuckets = new HashMap<>(); List simpleFileEntries = @@ -54,6 +109,18 @@ public static Map getKnownNumBuckets(FileStoreTable table) { return knownNumBuckets; } + /** Returns row counts of current active files in the postpone bucket. */ + public static Map getPostponeRowCounts(FileStoreTable table) { + Map rowCounts = new HashMap<>(); + Iterator 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 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 000000000000..50d9bc207785 --- /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 knownNumBuckets = new HashMap<>(); + Map 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 7c4676397523..cbdcf825f2e2 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.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 targetRowNumPerBucket = + options.getOptional(CoreOptions.POSTPONE_TARGET_ROW_NUM_PER_BUCKET); + Map knownNumBuckets = PostponeUtils.getKnownNumBuckets(table); + Map 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 bucketOptions = new HashMap<>(table.options()); @@ -322,16 +329,13 @@ protected boolean buildForPostponeBucketCompaction( String commitUser = CoreOptions.createCommitUser(options); List> dataStreams = new ArrayList<>(); for (BinaryRow partition : partitions) { - int bucketNum = defaultBucketNum; - - Iterator 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 d285fbb0449c..d9f8781da3b2 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 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 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 d714f47541a1..4725ce09f91e 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 d984f1385cb2..1bc3cc188834 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("""