[spark] Infer postpone bucket count from data size

[JingsongLi]

What changed

• Infer fixed bucket counts for new postpone partitions from the exact uncompressed Paimon BinaryRow size of incoming rows.
• Add postpone.target-size-per-bucket, defaulting to 1 GB.
• Estimate existing postpone data with its row count and the average incoming serialized row size for the same partition.
• Keep postpone.target-row-num-per-bucket as an explicit row-count override.
• Preserve the known bucket count for partitions that already have real buckets.
• Cache Spark input between the statistics pass and the write when inference is required.
• Skip the statistics pass and cache for an unpartitioned table with real buckets, or when postpone.batch-write-fixed-bucket.max-parallelism is 1.
• Exclude existing postpone rows when a full, static-partition, or dynamic-partition overwrite replaces them.
• Cap inferred counts with postpone.batch-write-fixed-bucket.max-parallelism before converting row-based counts to Int.
• Document the behavior and add Spark regression coverage.

Why

Spark previously used the number of input writers as the bucket count for partitions without real bucket data. Highly parallel jobs could therefore initialize many mostly empty buckets and create excessive small files, regardless of the actual data volume.

The serialized size is measured before ORC or Parquet encoding and compression, so the target represents logical uncompressed data rather than exact storage bytes.

Validation

• mvn -pl paimon-spark/paimon-spark-ut -am -Pfast-build -Pspark3 -DfailIfNoTests=false -DwildcardSuites=org.apache.paimon.spark.sql.PostponeBucketTableTest -Dtest=none test (20 tests passed)
• mvn -pl paimon-spark/paimon-spark-common,paimon-spark/paimon-spark-ut -Pspark3 -DskipTests validate
• mvn -pl paimon-spark/paimon-spark-common -am -Pspark3 -DskipTests compile
• mvn package -Pgenerate-docs -pl paimon-docs -nsu -DskipTests -am

[leaves12138]

I found two correctness issues in the new bucket inference path. The existing PostponeBucketTableTest passes (18/18), and targeted cache-reuse and mixed-RowKind tests also pass. The overwrite case below is reproducible with the PR source.

[leaves12138]

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.

[leaves12138]

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.

[leaves12138]

Re-reviewed commit c466234. Both previously reported correctness issues are fixed: overwrite inference now excludes postpone rows that will be removed, and row-count inference caps before Int conversion. I also verified all 20 PostponeBucketTableTest cases with Spark 3 and compiled the changed Spark common module with Spark 4 / Scala 2.13. No further blocking issues found.