[core][flink][spark] Support default bucket number for postpone writes

docs/docs/primary-key-table/data-distribution.md

@@ -70,16 +70,18 @@ Postpone bucket mode is configured by `'bucket' = '-2'`.
 This mode aims to solve the difficulty to determine a fixed number of buckets
 and support different buckets for different partitions.
 
-By default, `postpone.batch-write-fixed-bucket` is `true`. This staged fixed-bucket flow uses
-Spark's DataSource V1 write path, even when `spark.paimon.write.use-v2-write` is enabled. Spark
-completes each batch in three steps:
+By default, `postpone.batch-write-fixed-bucket` is `true`. The fixed-bucket flow uses Spark's
+DataSource V1 write path, even when `spark.paimon.write.use-v2-write` is enabled. Unless direct
+writing applies, Spark completes each batch in three steps:
 
 1. Write the current batch to uncommitted bucket `-2` files. Spark derives each partition's row
    count and file size directly from the staged file metadata; there is no extra input scan, cache,
    or per-row statistics pass.
-2. Calculate the required bucket number per touched partition. `postpone.target-row-num-per-bucket`,
-   when configured, takes precedence over `postpone.target-size-per-bucket` (default `1 GB`). The
-   result is at least `1`, rounded up to a power of two, and capped by
+2. Calculate the required bucket number per touched partition. For a partition without real
+   buckets, an explicitly configured `postpone.default-bucket-num` is used exactly. Otherwise,
+   `postpone.target-row-num-per-bucket`, when configured, takes precedence over
+   `postpone.target-size-per-bucket` (default `1 GB`). An inferred result is at least `1`, rounded
+   up to a power of two, and capped by
    `postpone.batch-write-fixed-bucket.max-parallelism`.
 3. Route the staged records to real buckets and commit them. The current batch becomes visible only
    in this commit.
@@ -91,6 +93,14 @@ larger than the existing layout. Different partitions may have different target
 The rescale is a separate overwrite commit which changes real buckets only; the current batch is
 appended in the following commit.
 
+`postpone.default-bucket-num` has no default value. When it is explicitly configured, Spark can
+skip the staged bucket `-2` files and write directly to real buckets for `INSERT OVERWRITE`, or
+when the base snapshot contains no real buckets. An overwrite always uses the configured number
+exactly and does not rescale the replaced layout. An append to an existing real-bucket partition
+ignores this option and still uses the staged batch to decide whether rescaling is required. If a
+batch mixes existing and new real-bucket partitions, the whole batch remains staged; only the new
+partitions use the configured default.
+
 Previously committed bucket `-2` files are not included in the calculation, read, rewritten, or
 deleted by an append or rescale. They remain available to merge-on-read and regular postpone
 compaction. `INSERT OVERWRITE` still follows its normal replacement semantics.
@@ -100,14 +110,13 @@ records are first stored in the `bucket-postpone` directory of each partition
 and are not available to readers.
 To move these records into the correct bucket and make them readable, run a compaction job.
 See `compact` [procedure](../flink/procedures).
-The bucket number for the partitions compacted for the first time
-is configured by the option `postpone.default-bucket-num`, whose default value is `1`.
-You can also configure `postpone.target-row-num-per-bucket` to calculate the bucket number
-from the row count of the files in the postpone bucket directory.
-The calculated bucket number is `ceil(row_count / postpone.target-row-num-per-bucket)`,
-and is at least `1`.
-When this option is configured, it takes precedence over `postpone.default-bucket-num`
-for partitions compacted for the first time.
+The bucket number for partitions compacted for the first time can be configured by the option
+`postpone.default-bucket-num`. Its value is used exactly and takes precedence over automatic
+estimation. Otherwise, `postpone.target-row-num-per-bucket`, when configured, calculates the
+bucket number as `ceil(row_count / target_row_count)`. If it is not configured, Paimon calculates
+the bucket number as `ceil(postpone_file_size / postpone.target-size-per-bucket)`; the target size
+defaults to `1 GB`. Both estimates are at least `1`. Execution parallelism does not determine the
+logical bucket number.
 Partitions that already have real bucket files keep their existing bucket number.
 
 Finally, when you feel that the bucket number of some partition is too small,

docs/generated/core_configuration.html

@@ -1316,9 +1316,9 @@
         </tr>
         <tr>
             <td><h5>postpone.default-bucket-num</h5></td>
-            <td style="word-wrap: break-word;">1</td>
+            <td style="word-wrap: break-word;">(none)</td>
             <td>Integer</td>
-            <td>Bucket number for the partitions compacted for the first time in postpone bucket tables.</td>
+            <td>Optional bucket number for partitions receiving real buckets for the first time and for fixed-bucket overwrite writes. The configured value is used exactly and takes precedence over automatic bucket estimation. When unset, Paimon estimates the bucket number from the target row count or target file size.</td>
         </tr>
         <tr>
             <td><h5>postpone.merge-on-read</h5></td>
@@ -1330,13 +1330,13 @@
             <td><h5>postpone.target-row-num-per-bucket</h5></td>
             <td style="word-wrap: break-word;">(none)</td>
             <td>Long</td>
-            <td>Target row number per bucket when estimating the required bucket number from the current staged batch or compacting postpone bucket files.</td>
+            <td>Target postpone row count per bucket when estimating the required bucket number from staged or committed postpone files. When configured, this option takes precedence over 'postpone.target-size-per-bucket'.</td>
         </tr>
         <tr>
             <td><h5>postpone.target-size-per-bucket</h5></td>
             <td style="word-wrap: break-word;">1 gb</td>
             <td>MemorySize</td>
-            <td>Target staged file size per bucket when Spark estimates the required bucket number from the current staged batch. This option is ignored when 'postpone.target-row-num-per-bucket' is configured.</td>
+            <td>Target postpone file size per bucket when estimating the required bucket number from staged or committed postpone files. This option is ignored when 'postpone.target-row-num-per-bucket' is configured.</td>
         </tr>
         <tr>
             <td><h5>primary-key</h5></td>

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

@@ -2766,23 +2766,23 @@ public String toString() {
     public static final ConfigOption<Integer> POSTPONE_DEFAULT_BUCKET_NUM =
             key("postpone.default-bucket-num")
                     .intType()
-                    .defaultValue(1)
+                    .noDefaultValue()
                     .withDescription(
-                            "Bucket number for the partitions compacted for the first time in postpone bucket tables.");
+                            "Optional bucket number for partitions receiving real buckets for the first time and for fixed-bucket overwrite writes. The configured value is used exactly and takes precedence over automatic bucket estimation. When unset, Paimon estimates the bucket number from the target row count or target file size.");
 
     public static final ConfigOption<Long> POSTPONE_TARGET_ROW_NUM_PER_BUCKET =
             key("postpone.target-row-num-per-bucket")
                     .longType()
                     .noDefaultValue()
                     .withDescription(
-                            "Target row number per bucket when estimating the required bucket number from the current staged batch or compacting postpone bucket files.");
+                            "Target postpone row count per bucket when estimating the required bucket number from staged or committed postpone files. When configured, this option takes precedence over 'postpone.target-size-per-bucket'.");
 
     public static final ConfigOption<MemorySize> POSTPONE_TARGET_SIZE_PER_BUCKET =
             key("postpone.target-size-per-bucket")
                     .memoryType()
                     .defaultValue(MemorySize.parse("1 gb"))
                     .withDescription(
-                            "Target staged file size per bucket when Spark estimates the required bucket number from the current staged batch. "
+                            "Target postpone file size per bucket when estimating the required bucket number from staged or committed postpone files. "
                                     + "This option is ignored when 'postpone.target-row-num-per-bucket' is configured.");
 
     public static final ConfigOption<Long> GLOBAL_INDEX_ROW_COUNT_PER_SHARD =
@@ -4465,8 +4465,15 @@ public int postponeBatchWriteFixedBucketRescaleLoadFactor() {
         return options.get(POSTPONE_BATCH_WRITE_FIXED_BUCKET_RESCALE_LOAD_FACTOR);
     }
 
-    public int postponeDefaultBucketNum() {
-        return options.get(POSTPONE_DEFAULT_BUCKET_NUM);
+    public Optional<Integer> postponeDefaultBucketNum() {
+        Optional<Integer> bucketNum = options.getOptional(POSTPONE_DEFAULT_BUCKET_NUM);
+        bucketNum.ifPresent(
+                value ->
+                        checkArgument(
+                                value > 0,
+                                "Option '%s' must be greater than 0.",
+                                POSTPONE_DEFAULT_BUCKET_NUM.key()));
+        return bucketNum;
     }
 
     public Optional<Long> postponeTargetRowNumPerBucket() {