You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JingsongLi (via GitHub)" <gi...@apache.org> on 2026/07/27 13:31:13 UTC
[PR] [spark] Infer postpone bucket count from row count [paimon]
JingsongLi opened a new pull request, #8864:
URL: https://github.com/apache/paimon/pull/8864
## What changed
- Infer the fixed bucket count for new postpone partitions from exact incoming row counts when `postpone.target-row-num-per-bucket` is configured.
- Include existing postpone file row counts and cap the inferred count with `postpone.batch-write-fixed-bucket.max-parallelism`.
- Preserve known bucket counts and the existing writer-count fallback when the target row count is absent.
- Cache the Spark input between counting and writing.
- Document the behavior and add Spark regression coverage.
## Why
Spark previously used the number of input writers as the bucket count for partitions without real bucket data. Highly parallel jobs could therefore initialize many mostly empty buckets and create excessive small files.
## Validation
- `mvn -pl paimon-spark/paimon-spark-ut -am -Pfast-build -Pspark3 -DfailIfNoTests=false -DwildcardSuites=org.apache.paimon.spark.sql.PostponeBucketTableTest -Dtest=none test`
- `mvn -pl paimon-api,paimon-spark/paimon-spark-common,paimon-spark/paimon-spark-ut -Pspark3 -DskipTests compile`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Infer postpone bucket count from data size [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #8864:
URL: https://github.com/apache/paimon/pull/8864#discussion_r3658641898
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -536,6 +544,130 @@ case class PaimonSparkWriter(
.toSeq
}
+ private def preparePostponeBucketAssignment(df: DataFrame): 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 = PostponeUtils.getPostponeRowCounts(table)
+ val dataStats =
+ collectDataStatsByPartition(df, collectSize = !targetRowNum.isPresent)
+ dataStats.map {
+ case (partition, stats) =>
+ val postponeRowCount = postponeRowCounts.getOrDefault(partition, 0L)
Review Comment:
Existing postpone rows must not be included for an overwrite. `WriteIntoPaimonTable` configures `writeBuilder.withOverwrite(...)` before calling `write`, so the postpone files in the overwritten partition are removed by this commit. Counting them inflates the bucket number for data that will no longer exist. I reproduced this with 1,000 postpone rows followed by `INSERT OVERWRITE` of 100 rows and `postpone.target-row-num-per-bucket = 100`: the final partition is written with 11 buckets instead of 1. Please pass the overwrite semantics into inference (including dynamic partition overwrite) and exclude postpone rows that the commit removes.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -536,6 +544,130 @@ case class PaimonSparkWriter(
.toSeq
}
+ private def preparePostponeBucketAssignment(df: DataFrame): 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 = 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) {
+ Math.min(
+ PostponeUtils.computeBucketNumByRowCount(
Review Comment:
The maximum is applied only after `computeBucketNumByRowCount`, but that helper throws when the computed value exceeds `Integer.MAX_VALUE`. For example, 2,147,483,648 rows with a target of 1 and a max parallelism of 2,048 fail instead of being capped at 2,048, contradicting the documented limit. Please compute/cap in `Long` or `BigInt` before converting to `Int`, analogous to the size-based path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Infer postpone bucket count from data size [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #8864:
URL: https://github.com/apache/paimon/pull/8864#discussion_r3658641898
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -536,6 +544,130 @@ case class PaimonSparkWriter(
.toSeq
}
+ private def preparePostponeBucketAssignment(df: DataFrame): 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 = PostponeUtils.getPostponeRowCounts(table)
+ val dataStats =
+ collectDataStatsByPartition(df, collectSize = !targetRowNum.isPresent)
+ dataStats.map {
+ case (partition, stats) =>
+ val postponeRowCount = postponeRowCounts.getOrDefault(partition, 0L)
Review Comment:
Existing postpone rows must not be included for an overwrite. configures before calling , so the postpone files in the overwritten partition are removed by this commit. Counting them inflates the bucket number for data that will no longer exist. I reproduced this with 1,000 postpone rows followed by of 100 rows and : the final partition is written with 11 buckets instead of 1. Please pass the overwrite semantics into inference (including dynamic partition overwrite) and exclude postpone rows that the commit removes.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala:
##########
@@ -536,6 +544,130 @@ case class PaimonSparkWriter(
.toSeq
}
+ private def preparePostponeBucketAssignment(df: DataFrame): 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 = 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) {
+ Math.min(
+ PostponeUtils.computeBucketNumByRowCount(
Review Comment:
The maximum is applied only after , but that helper throws when the computed value exceeds . For example, 2,147,483,648 rows with a target of 1 and a max parallelism of 2,048 fail instead of being capped at 2,048, contradicting the documented limit. Please compute/cap in or before converting to , analogous to the size-based path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [spark] Infer postpone bucket count from data size [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #8864:
URL: https://github.com/apache/paimon/pull/8864
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org