(paimon) branch master updated: [spark] Recompute bucket number for postpone overwrite (#9051)

[lzljs3620320]

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 7d10cf2178 [spark] Recompute bucket number for postpone overwrite 
(#9051)
7d10cf2178 is described below

commit 7d10cf2178f14cc6fbe0d6cc6b309b752ec7b745
Author: Zouxxyy <zouxinyu....@alibaba-inc.com>
AuthorDate: Thu Aug 6 09:03:22 2026 +0800

    [spark] Recompute bucket number for postpone overwrite (#9051)
---
 .../spark/SparkPostponeStagedCommitter.scala       | 27 +++++++++++-------
 .../paimon/spark/sql/PostponeBucketTableTest.scala | 33 ++++++++++++++--------
 2 files changed, 39 insertions(+), 21 deletions(-)

diff --git 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
index 1ba596b4fe..883a797ebe 100644
--- 
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
+++ 
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala
@@ -105,16 +105,23 @@ private[spark] class SparkPostponeStagedCommitter(
         return passThroughMessages
       }
 
-      val existingBuckets: Map[BinaryRow, Int] = baseSnapshotId
-        .map(
-          id =>
-            PostponeUtils
-              .getKnownNumBuckets(table, id, touchedPartitions.asJava)
-              .asScala
-              .iterator
-              .map { case (partition, buckets) => partition -> 
buckets.intValue() }
-              .toMap)
-        .getOrElse(Map.empty[BinaryRow, Int])
+      // Overwrite removes the previous partition contents, so its bucket 
layout must not
+      // constrain the replacement data. It is also unnecessary to scan the 
old layout here.
+      val existingBuckets: Map[BinaryRow, Int] =
+        if (overwritePartitionSpec.isDefined) {
+          Map.empty
+        } else {
+          baseSnapshotId
+            .map(
+              id =>
+                PostponeUtils
+                  .getKnownNumBuckets(table, id, touchedPartitions.asJava)
+                  .asScala
+                  .iterator
+                  .map { case (partition, buckets) => partition -> 
buckets.intValue() }
+                  .toMap)
+            .getOrElse(Map.empty[BinaryRow, Int])
+        }
       val decisions = touchedPartitions.map {
         partition =>
           val stage = stats(partition)
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 a96e0c21e9..2f59dd9e0c 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
@@ -643,7 +643,7 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
     }
   }
 
-  test("Postpone bucket table: overwrite excludes existing postpone rows from 
inference") {
+  test("Postpone bucket table: overwrite recomputes bucket number from staged 
batch") {
     Seq(
       (
         "static",
@@ -690,13 +690,23 @@ class PostponeBucketTableTest extends PaimonSparkTestBase 
{
                   |""".stripMargin)
             assert(SparkTable(loadTable("t")).useV2Write)
 
-            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") {
+              sql("""
+                    |INSERT INTO t SELECT
+                    |id AS k,
+                    |CAST(id AS STRING) AS v,
+                    |0 AS pt
+                    |FROM range (0, 1000)
+                    |""".stripMargin)
+            }
+            assert(
+              PostponeUtils
+                .getKnownNumBuckets(loadTable("t"))
+                .get(BinaryRow.singleColumn(0)) == 16)
+
+            withSparkSQLConf("spark.paimon.postpone.batch-write-fixed-bucket" 
-> "false") {
+              sql("INSERT INTO t VALUES (2000, 'historical-postpone', 0)")
+            }
 
             withSparkSQLConf(
               "spark.paimon.postpone.batch-write-fixed-bucket" -> "true",
@@ -705,9 +715,10 @@ class PostponeBucketTableTest extends PaimonSparkTestBase {
             }
 
             checkAnswer(sql("SELECT count(*), sum(k) FROM t"), Seq(Row(100L, 
104950L)))
-            checkAnswer(
-              sql("SELECT distinct(bucket) FROM `t$buckets` WHERE partition = 
'{0}'"),
-              Seq(Row(0)))
+            assert(
+              PostponeUtils
+                .getKnownNumBuckets(loadTable("t"))
+                .get(BinaryRow.singleColumn(0)) == 1)
             checkAnswer(sql("SELECT count(*) FROM `t$buckets` WHERE bucket = 
-2"), Seq(Row(0L)))
           }
         }