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