[core] Limit max parallelism for postpone batch write fixed bucket to avoid too many buckets (#7212)

Author: yuzelin

docs/layouts/shortcodes/generated/core_configuration.html

@@ -152,12 +152,6 @@
             <td>String</td>
             <td>Fields that are ignored for comparison while generating -U, +U changelog for the same record. This configuration is only valid for the changelog-producer.row-deduplicate is true.</td>
         </tr>
-        <tr>
-            <td><h5>table-read.sequence-number.enabled</h5></td>
-            <td style="word-wrap: break-word;">false</td>
-            <td>Boolean</td>
-            <td>Whether to include the _SEQUENCE_NUMBER field when reading the audit_log or binlog system tables. This is only valid for primary key tables.</td>
-        </tr>
         <tr>
             <td><h5>changelog.num-retained.max</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
@@ -1031,6 +1025,12 @@
             <td>Boolean</td>
             <td>Whether to write the data into fixed bucket for batch writing a postpone bucket table.</td>
         </tr>
+        <tr>
+            <td><h5>postpone.batch-write-fixed-bucket.max-parallelism</h5></td>
+            <td style="word-wrap: break-word;">2048</td>
+            <td>Integer</td>
+            <td>The number of partitions for global index.</td>
+        </tr>
         <tr>
             <td><h5>postpone.default-bucket-num</h5></td>
             <td style="word-wrap: break-word;">1</td>
@@ -1314,6 +1314,12 @@
             <td>Duration</td>
             <td>The delay duration of stream read when scan incremental snapshots.</td>
         </tr>
+        <tr>
+            <td><h5>table-read.sequence-number.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to include the _SEQUENCE_NUMBER field when reading the audit_log or binlog system tables. This is only valid for primary key tables.</td>
+        </tr>
         <tr>
             <td><h5>tag.automatic-completion</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2164,6 +2164,12 @@ public InlineElement getDescription() {
                     .withDescription(
                             "Whether to write the data into fixed bucket for batch writing a postpone bucket table.");
 
+    public static final ConfigOption<Integer> POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM =
+            key("postpone.batch-write-fixed-bucket.max-parallelism")
+                    .intType()
+                    .defaultValue(2048)
+                    .withDescription("The number of partitions for global index.");
+
     public static final ConfigOption<Integer> POSTPONE_DEFAULT_BUCKET_NUM =
             key("postpone.default-bucket-num")
                     .intType()
@@ -3401,6 +3407,10 @@ public boolean postponeBatchWriteFixedBucket() {
         return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET);
     }
 
+    public int postponeBatchWriteFixedBucketMaxParallelism() {
+        return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_MAX_PARALLELISM);
+    }
+
     public int postponeDefaultBucketNum() {
         return options.get(POSTPONE_DEFAULT_BUCKET_NUM);
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeBatchWriteOperator.java

@@ -72,7 +72,11 @@ public void open() throws Exception {
         super.open();
 
         int sinkParallelism = RuntimeContextUtils.getNumberOfParallelSubtasks(getRuntimeContext());
-        this.defaultNumBuckets = sinkParallelism <= 0 ? 1 : sinkParallelism;
+        sinkParallelism = sinkParallelism <= 0 ? 1 : sinkParallelism;
+        this.defaultNumBuckets =
+                Math.min(
+                        sinkParallelism,
+                        table.coreOptions().postponeBatchWriteFixedBucketMaxParallelism());
 
         TableSchema schema = table.schema();
         this.partitionKeyExtractor = new RowPartitionKeyExtractor(schema);

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -123,7 +123,9 @@ case class PaimonSparkWriter(
     val postponePartitionBucketComputer: Option[BinaryRow => Integer] =
       if (postponeBatchWriteFixedBucket) {
         val knownNumBuckets = PostponeUtils.getKnownNumBuckets(table)
-        val defaultPostponeNumBuckets = withInitBucketCol.rdd.getNumPartitions
+        val defaultPostponeNumBuckets = Math.min(
+          withInitBucketCol.rdd.getNumPartitions,
+          table.coreOptions().postponeBatchWriteFixedBucketMaxParallelism)
         Some((p: BinaryRow) => knownNumBuckets.getOrDefault(p, defaultPostponeNumBuckets))
       } else {
         None