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 9ad20d9978 [spark] Infer postpone bucket count from data size (#8864)
9ad20d9978 is described below
commit 9ad20d997858c124ae33654f6a1fa694dc588211
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 28 12:07:45 2026 +0800
[spark] Infer postpone bucket count from data size (#8864)
---
docs/docs/primary-key-table/data-distribution.md | 25 ++-
docs/generated/core_configuration.html | 8 +-
.../main/java/org/apache/paimon/CoreOptions.java | 14 +-
.../paimon/spark/commands/PaimonSparkWriter.scala | 196 +++++++++++++++++-
.../spark/commands/WriteIntoPaimonTable.scala | 5 +-
.../paimon/spark/sql/PostponeBucketTableTest.scala | 219 ++++++++++++++++++++-
6 files changed, 442 insertions(+), 25 deletions(-)
diff --git a/docs/docs/primary-key-table/data-distribution.md
b/docs/docs/primary-key-table/data-distribution.md
index 49846b2ea3..a7cf7aa16d 100644
--- a/docs/docs/primary-key-table/data-distribution.md
+++ b/docs/docs/primary-key-table/data-distribution.md
@@ -70,12 +70,27 @@ Postpone bucket mode is configured by `'bucket' = '-2'`.
This mode aims to solve the difficulty to determine a fixed number of buckets
and support different buckets for different partitions.
-When writing records into the table,
-all records will first be stored in the `bucket-postpone` directory of each
partition
+By default, batch writes set `postpone.batch-write-fixed-bucket` to `true`
+and write records directly to real buckets.
+For a Spark batch write to a partition without real bucket files,
+Spark calculates the bucket number from the uncompressed Paimon `BinaryRow`
serialized size.
+The target size is configured by `postpone.target-size-per-bucket` and
defaults to `1 GB`.
+Existing postpone files do not record their uncompressed serialized size, so
Spark estimates their
+size using their row count and the average serialized size of incoming rows in
the same partition.
+You can instead configure `postpone.target-row-num-per-bucket` to calculate
the bucket number
+from row counts; this option takes precedence over the target size.
+The calculated bucket number is at least `1` and is limited by
+`postpone.batch-write-fixed-bucket.max-parallelism`.
+The serialized size is measured before file encoding and compression, so it is
not the exact
+ORC or Parquet size on storage.
+Inferring the data amount adds a statistics stage; Spark caches the batch so
that inference and
+writing use the same input rows. Spark skips this stage for an unpartitioned
table that already
+has real bucket files, or when
`postpone.batch-write-fixed-bucket.max-parallelism` is `1`.
+
+When `postpone.batch-write-fixed-bucket` is `false`,
+records are first stored in the `bucket-postpone` directory of each partition
and are not available to readers.
-
-To move the records into the correct bucket and make them readable,
-you need to run a compaction job.
+To move these records into the correct bucket and make them readable, 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`.
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 4d1d8f87f7..7bc238d4ff 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1294,7 +1294,13 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<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>
+ <td>Target row number per bucket when batch writing fixed buckets
or compacting postpone bucket files for a partition without real bucket
data.</td>
+ </tr>
+ <tr>
+ <td><h5>postpone.target-size-per-bucket</h5></td>
+ <td style="word-wrap: break-word;">1 gb</td>
+ <td>MemorySize</td>
+ <td>Target uncompressed serialized data size per bucket when Spark
batch writes fixed buckets for a partition without real bucket data. This
option is ignored when 'postpone.target-row-num-per-bucket' is configured.</td>
</tr>
<tr>
<td><h5>primary-key</h5></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 3040a981a4..ec95ab6ec5 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2733,7 +2733,15 @@ public class CoreOptions implements Serializable {
.longType()
.noDefaultValue()
.withDescription(
- "Target row number per bucket for partitions
compacted from postpone bucket files for the first time.");
+ "Target row number per bucket when batch writing
fixed buckets or compacting postpone bucket files for a partition without real
bucket data.");
+
+ public static final ConfigOption<MemorySize>
POSTPONE_TARGET_SIZE_PER_BUCKET =
+ key("postpone.target-size-per-bucket")
+ .memoryType()
+ .defaultValue(MemorySize.parse("1 gb"))
+ .withDescription(
+ "Target uncompressed serialized data size per
bucket when Spark batch writes fixed buckets for a partition without real
bucket data. "
+ + "This option is ignored when
'postpone.target-row-num-per-bucket' is configured.");
public static final ConfigOption<Long> GLOBAL_INDEX_ROW_COUNT_PER_SHARD =
key("global-index.row-count-per-shard")
@@ -4406,6 +4414,10 @@ public class CoreOptions implements Serializable {
return options.getOptional(POSTPONE_TARGET_ROW_NUM_PER_BUCKET);
}
+ public long postponeTargetSizePerBucket() {
+ return options.get(POSTPONE_TARGET_SIZE_PER_BUCKET).getBytes();
+ }
+
public long globalIndexRowCountPerShard() {
return options.get(GLOBAL_INDEX_ROW_COUNT_PER_SHARD);
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 1518a77934..aaba0f1ca0 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -52,6 +52,7 @@ import java.io.IOException
import java.util.Collections.singletonMap
import scala.collection.JavaConverters._
+import scala.collection.mutable
case class PaimonSparkWriter(
table: FileStoreTable,
@@ -59,6 +60,8 @@ case class PaimonSparkWriter(
batchId: Option[Long] = None)
extends WriteHelper {
+ import PaimonSparkWriter._
+
private lazy val tableSchema = table.schema
private lazy val bucketMode = table.bucketMode
@@ -106,6 +109,12 @@ case class PaimonSparkWriter(
}
def write(data: DataFrame): Seq[CommitMessage] = {
+ write(data, overwriteExistingData = false)
+ }
+
+ private[commands] def write(
+ data: DataFrame,
+ overwriteExistingData: Boolean): Seq[CommitMessage] = {
val sparkSession = data.sparkSession
val uriReaderFactory = uriReaderFactoryForBlobDescriptor
import sparkSession.implicits._
@@ -118,20 +127,19 @@ case class PaimonSparkWriter(
.withColumn(BUCKET_COL, lit(-1))
case _ => data.withColumn(BUCKET_COL, lit(-1))
}
- val rowKindColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema,
ROW_KIND_COL)
- val bucketColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema,
BUCKET_COL)
- val encoderGroupWithBucketCol = EncoderSerDeGroup(withInitBucketCol.schema)
- val postponePartitionBucketComputer: Option[BinaryRow => Integer] =
+ val postponeBucketAssignment =
if (postponeBatchWriteFixedBucket) {
- val knownNumBuckets = PostponeUtils.getKnownNumBuckets(table)
- val defaultPostponeNumBuckets = Math.min(
- withInitBucketCol.rdd.getNumPartitions,
- table.coreOptions().postponeBatchWriteFixedBucketMaxParallelism)
- Some((p: BinaryRow) => knownNumBuckets.getOrDefault(p,
defaultPostponeNumBuckets))
+ Some(preparePostponeBucketAssignment(withInitBucketCol,
overwriteExistingData))
} else {
None
}
+ val rowKindColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema,
ROW_KIND_COL)
+ val bucketColIdx = SparkRowUtils.getFieldIndex(withInitBucketCol.schema,
BUCKET_COL)
+ val encoderGroupWithBucketCol = EncoderSerDeGroup(withInitBucketCol.schema)
+ val postponePartitionBucketComputer =
+ postponeBucketAssignment.map(_.partitionBucketComputer)
+
def newWrite() = PaimonDataWrite(
writeBuilder,
writeType,
@@ -349,7 +357,13 @@ case class PaimonSparkWriter(
throw new UnsupportedOperationException(s"Spark doesn't support
$bucketMode mode.")
}
- WriteTaskResult.merge(written.collect())
+ try {
+ WriteTaskResult.merge(written.collect())
+ } finally {
+ if (postponeBucketAssignment.exists(_.dataPersisted)) {
+ withInitBucketCol.unpersist()
+ }
+ }
}
/**
@@ -536,6 +550,136 @@ case class PaimonSparkWriter(
.toSeq
}
+ private def preparePostponeBucketAssignment(
+ df: DataFrame,
+ overwriteExistingData: Boolean): PostponeBucketAssignment = {
+ val knownNumBuckets = PostponeUtils.getKnownNumBuckets(table)
+ val maxNumBuckets =
coreOptions.postponeBatchWriteFixedBucketMaxParallelism()
+ val unpartitionedTableHasKnownNumBuckets =
+ tableSchema.partitionKeys().isEmpty &&
+ knownNumBuckets.containsKey(BinaryRow.EMPTY_ROW)
+ val inferBucketNumFromData =
+ maxNumBuckets != 1 && !unpartitionedTableHasKnownNumBuckets
+ if (inferBucketNumFromData) {
+ df.persist()
+ }
+
+ try {
+ val defaultNumBuckets = Math.min(df.rdd.getNumPartitions, maxNumBuckets)
+ val inferredNumBuckets: Map[BinaryRow, Int] =
+ if (inferBucketNumFromData) {
+ val targetRowNum = coreOptions.postponeTargetRowNumPerBucket()
+ val postponeRowCounts =
+ if (overwriteExistingData) {
+ java.util.Collections.emptyMap[BinaryRow, java.lang.Long]()
+ } else {
+ PostponeUtils.getPostponeRowCounts(table)
+ }
+ val dataStats =
+ collectDataStatsByPartition(df, collectSize =
!targetRowNum.isPresent)
+ dataStats.map {
+ case (partition, stats) =>
+ val postponeRowCount = postponeRowCounts.getOrDefault(partition,
0L)
+ val numBuckets =
+ if (targetRowNum.isPresent) {
+ computeBucketNumByRowCount(
+ Math.addExact(stats.rowCount, postponeRowCount),
+ targetRowNum.get(),
+ maxNumBuckets)
+ } else {
+ computeBucketNumBySize(
+ stats,
+ postponeRowCount,
+ coreOptions.postponeTargetSizePerBucket(),
+ maxNumBuckets)
+ }
+ partition -> numBuckets
+ }
+ } else {
+ Map.empty
+ }
+ val partitionBucketComputer = (partition: BinaryRow) =>
+ knownNumBuckets.getOrDefault(
+ partition,
+ Integer.valueOf(inferredNumBuckets.getOrElse(partition,
defaultNumBuckets)))
+ PostponeBucketAssignment(partitionBucketComputer, inferBucketNumFromData)
+ } catch {
+ case e: Throwable =>
+ if (inferBucketNumFromData) {
+ df.unpersist()
+ }
+ throw e
+ }
+ }
+
+ private def collectDataStatsByPartition(
+ df: DataFrame,
+ collectSize: Boolean): Map[BinaryRow, PartitionDataStats] = {
+ val schema = tableSchema
+ val rowType = writeType
+ val toPaimonRow = SparkRowUtils.toPaimonRow(
+ rowType,
+ SparkRowUtils.getFieldIndex(df.schema, ROW_KIND_COL),
+ uriReaderFactoryForBlobDescriptor)
+ df.rdd
+ .mapPartitions {
+ rows =>
+ val partitionKeyExtractor = new RowPartitionKeyExtractor(schema)
+ val rowSerializer = InternalSerializers.create(rowType)
+ val stats = mutable.HashMap.empty[SerializedPartition,
PartitionDataStats]
+ rows.foreach {
+ row =>
+ val paimonRow = toPaimonRow(row)
+ val partition = SerializedPartition(
+
SerializationUtils.serializeBinaryRow(partitionKeyExtractor.partition(paimonRow)))
+ val rowSize =
+ if (collectSize)
rowSerializer.toBinaryRow(paimonRow).getSizeInBytes.toLong else 0L
+ val previous = stats.getOrElse(partition, PartitionDataStats(0L,
0L))
+ stats.put(
+ partition,
+ PartitionDataStats(
+ Math.addExact(previous.rowCount, 1L),
+ Math.addExact(previous.serializedSize, rowSize)))
+ }
+ stats.iterator
+ }
+ .reduceByKey {
+ (left, right) =>
+ PartitionDataStats(
+ Math.addExact(left.rowCount, right.rowCount),
+ Math.addExact(left.serializedSize, right.serializedSize))
+ }
+ .collect()
+ .map {
+ case (partition, stats) =>
+ SerializationUtils.deserializeBinaryRow(partition.bytes) -> stats
+ }
+ .toMap
+ }
+
+ private def computeBucketNumBySize(
+ dataStats: PartitionDataStats,
+ postponeRowCount: Long,
+ targetSizePerBucket: Long,
+ maxNumBuckets: Int): Int = {
+ if (targetSizePerBucket <= 0) {
+ throw new IllegalArgumentException(
+ "Option 'postpone.target-size-per-bucket' must be greater than 0.")
+ }
+
+ // Previous postpone files do not record their uncompressed serialized
size. Estimate it with
+ // the average serialized size of incoming rows from the same partition.
+ val estimatedTotalSizeNumerator =
+ BigInt(dataStats.serializedSize) *
+ (BigInt(dataStats.rowCount) + BigInt(postponeRowCount))
+ val rowCount = BigInt(dataStats.rowCount)
+ val estimatedTotalSize = (estimatedTotalSizeNumerator + rowCount - 1) /
rowCount
+ val bucketNum =
+ if (estimatedTotalSize == 0) BigInt(1)
+ else (estimatedTotalSize - 1) / BigInt(targetSizePerBucket) + 1
+ bucketNum.min(BigInt(maxNumBuckets)).toInt
+ }
+
private def deserializeCommitMessage(
serializer: CommitMessageSerializer,
bytes: Array[Byte]): CommitMessage = {
@@ -551,10 +695,42 @@ case class PaimonSparkWriter(
override def numPartitions: Int = partitions
override def getPartition(key: Any): Int = Math.abs(key.asInstanceOf[Int]
% numPartitions)
}
+
}
object PaimonSparkWriter {
+ private[spark] def computeBucketNumByRowCount(
+ rowCount: Long,
+ targetRowNumPerBucket: Long,
+ maxNumBuckets: Int): Int = {
+ if (targetRowNumPerBucket <= 0) {
+ throw new IllegalArgumentException(
+ "Option 'postpone.target-row-num-per-bucket' must be greater than 0.")
+ }
+
+ val bucketNum =
+ if (rowCount <= 0) 1L else (rowCount - 1) / targetRowNumPerBucket + 1
+ Math.min(bucketNum, maxNumBuckets.toLong).toInt
+ }
+
+ private case class PostponeBucketAssignment(
+ partitionBucketComputer: BinaryRow => Integer,
+ dataPersisted: Boolean)
+
+ private case class PartitionDataStats(rowCount: Long, serializedSize: Long)
+
+ private case class SerializedPartition(bytes: Array[Byte]) {
+ override def equals(other: Any): Boolean = {
+ other match {
+ case that: SerializedPartition => java.util.Arrays.equals(bytes,
that.bytes)
+ case _ => false
+ }
+ }
+
+ override def hashCode(): Int = java.util.Arrays.hashCode(bytes)
+ }
+
def apply(table: FileStoreTable): PaimonSparkWriter = {
new PaimonSparkWriter(table)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
index f400409fda..346aa0fe3a 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
@@ -71,7 +71,10 @@ case class WriteIntoPaimonTable(
Snapshot.Operation.WRITE
}
}
- val commitMessages = writer.write(replacedData)
+ val commitMessages =
+ writer.write(
+ replacedData,
+ overwriteExistingData = overwritePartition != null ||
dynamicPartitionOverwriteMode)
writer.commit(commitMessages, operation)
Seq.empty
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 805b747bb6..1e9910bf7e 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
@@ -22,6 +22,7 @@ import org.apache.paimon.catalog.{Catalog, CatalogLoader,
DelegateCatalog, Ident
import
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX
import org.apache.paimon.fs.Path
import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase}
+import org.apache.paimon.spark.commands.PaimonSparkWriter
import org.apache.paimon.spark.procedure.SparkPostponeCompactProcedure
import org.apache.paimon.table.{CatalogEnvironment, FileStoreTableFactory}
@@ -33,6 +34,14 @@ import scala.collection.JavaConverters._
class PostponeBucketTableTest extends PaimonSparkTestBase {
+ test("Postpone bucket table: cap inferred row-count buckets before Int
conversion") {
+ assert(
+ PaimonSparkWriter.computeBucketNumByRowCount(
+ Integer.MAX_VALUE.toLong + 1L,
+ targetRowNumPerBucket = 1L,
+ maxNumBuckets = 2048) == 2048)
+ }
+
test("Postpone bucket table: write with different bucket number") {
withTable("t") {
sql("""
@@ -60,9 +69,11 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
checkAnswer(sql("SELECT sum(k) FROM t"), Seq(Row(499500)))
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2), Row(3))
+ Seq(Row(0))
)
+ sql("ALTER TABLE t SET TBLPROPERTIES ('postpone.target-size-per-bucket'
= '8 kb')")
+
// Write to existing partition, the bucket number should not change
sql("""
|INSERT INTO t SELECT /*+ REPARTITION(6) */
@@ -73,7 +84,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
|""".stripMargin)
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{3}'
ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2), Row(3))
+ Seq(Row(0))
)
// Write to new partition, the bucket number should change
@@ -86,7 +97,201 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
|""".stripMargin)
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{5}'
ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2), Row(3), Row(4), Row(5))
+ Seq(Row(0), Row(1), Row(2))
+ )
+ }
+ }
+
+ test("Postpone bucket table: infer bucket number from incoming row count") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-row-num-per-bucket' = '200'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |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 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))
+ )
+
+ // Existing partitions keep their bucket number even when the new data
volume changes.
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range (1000, 2000)
+ |""".stripMargin)
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{0}'
ORDER BY bucket"),
+ Seq(Row(0))
+ )
+
+ // Postpone rows are included when a partition gets real buckets for the
first time.
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
+ sql("""
+ |INSERT INTO t SELECT
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |2 AS pt
+ |FROM range (2000, 2150)
+ |""".stripMargin)
+ }
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |2 AS pt
+ |FROM range (3000, 3100)
+ |""".stripMargin)
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{2}'
ORDER BY bucket"),
+ Seq(Row(-2), Row(0), Row(1))
+ )
+ }
+ }
+
+ test("Postpone bucket table: overwrite excludes existing postpone rows from
inference") {
+ Seq(
+ (
+ "static",
+ """
+ |INSERT OVERWRITE t SELECT
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range (1000, 1100)
+ |""".stripMargin),
+ (
+ "static",
+ """
+ |INSERT OVERWRITE t PARTITION (pt = 0) SELECT
+ |id AS k,
+ |CAST(id AS STRING) AS v
+ |FROM range (1000, 1100)
+ |""".stripMargin),
+ (
+ "dynamic",
+ """
+ |INSERT OVERWRITE t SELECT
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range (1000, 1100)
+ |""".stripMargin)
+ ).foreach {
+ case (partitionOverwriteMode, overwriteSql) =>
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'false',
+ | 'postpone.target-row-num-per-bucket' = '100'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t SELECT
+ |id AS k,
+ |CAST(id AS STRING) AS v,
+ |0 AS pt
+ |FROM range (0, 1000)
+ |""".stripMargin)
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.batch-write-fixed-bucket" -> "true",
+ "spark.sql.sources.partitionOverwriteMode" ->
partitionOverwriteMode) {
+ sql(overwriteSql)
+ }
+
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition =
'{0}'"),
+ Seq(Row(0))
+ )
+ }
+ }
+ }
+
+ test("Postpone bucket table: infer bucket number from serialized data size")
{
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (
+ | k INT,
+ | v STRING,
+ | pt INT
+ |) PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'k, pt',
+ | 'bucket' = '-2',
+ | 'postpone.batch-write-fixed-bucket' = 'true',
+ | 'postpone.target-size-per-bucket' = '32 kb'
+ |)
+ |""".stripMargin)
+
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |id AS k,
+ |CASE WHEN id < 100 THEN repeat('x', 100) ELSE repeat('x', 1000)
END AS v,
+ |CASE WHEN id < 100 THEN 0 ELSE 1 END AS pt
+ |FROM range (0, 200)
+ |""".stripMargin)
+
+ 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), Row(3))
+ )
+
+ // Estimate existing postpone data with the average serialized size of
incoming rows.
+ withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" ->
"false") {
+ sql("""
+ |INSERT INTO t SELECT
+ |id AS k,
+ |repeat('x', 1000) AS v,
+ |2 AS pt
+ |FROM range (1000, 1100)
+ |""".stripMargin)
+ }
+ sql("""
+ |INSERT INTO t SELECT /*+ REPARTITION(20) */
+ |id AS k,
+ |repeat('x', 1000) AS v,
+ |2 AS pt
+ |FROM range (2000, 2100)
+ |""".stripMargin)
+ checkAnswer(
+ sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = '{2}'
ORDER BY bucket"),
+ Seq(Row(-2), Row(0), Row(1), Row(2), Row(3), Row(4), Row(5), Row(6))
)
}
}
@@ -116,7 +321,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
checkAnswer(sql("SELECT sum(k) FROM t"), Seq(Row(499500)))
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2), Row(3))
+ Seq(Row(0))
)
// write postpone bucket
@@ -131,7 +336,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
checkAnswer(sql("SELECT sum(k) FROM t"), Seq(Row(499500)))
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY bucket"),
- Seq(Row(-2), Row(0), Row(1), Row(2), Row(3))
+ Seq(Row(-2), Row(0))
)
}
}
@@ -605,7 +810,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
checkAnswer(sql("SELECT sum(k) FROM t"), Seq(Row(499500)))
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY bucket"),
- Seq(Row(-2), Row(0), Row(1), Row(2), Row(3), Row(4), Row(5))
+ Seq(Row(-2), Row(0))
)
}
@@ -621,7 +826,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
checkAnswer(sql("SELECT sum(k) FROM t"), Seq(Row(124750)))
checkAnswer(
sql("SELECT distinct(bucket) FROM `t$buckets` ORDER BY bucket"),
- Seq(Row(0), Row(1), Row(2), Row(3), Row(4), Row(5))
+ Seq(Row(0))
)
}
}
