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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
