[core][flink] Support Per-Partition Bucket Counts

[dwangatt]

Continuation of #7865

This PR continues #7865.

The original author is no longer available to maintain the source branch. This continuation preserves all 19 original commits and their authorship, and moves ongoing maintenance to the atlassian-forks/paimon fork.

Original PR:

• [core][flink] Support Per-Partition Bucket Counts #7865
• Original author: @mikedias
• Original branch: mdias/master/buckets-per-partition

Purpose

Support per-partition bucket counts so individual partitions can be rescaled independently without changing the bucket count of the entire table.

This allows a table to contain partitions with different bucket layouts while ensuring readers, writers, compaction, and overwrite/rescale operations use the bucket count recorded for each partition.

Commit history

All original commits from #7865 are preserved. New changes addressing outstanding review feedback will be added as separate commits on this maintained branch.

Follow-up

• Rebase and resolve conflicts against the latest apache/master
• Review and address unresolved feedback from [core][flink] Support Per-Partition Bucket Counts #7865
• Run the full unit and integration test suite
• Verify streaming writer behavior after a partition rescale
• Verify INSERT OVERWRITE/rescale behavior
• Verify writer coordinator and empty-bucket restore behavior

Supersedes #7865.

[dwangatt]

hey @JingsongLi following up the conversion from #7865
I added code so that:

• when bucket.per-partition-count-enabled is enabled, Paimon sink can use bucket from partitions so partitions can have different buckets
• when bucket.per-partition-count-enabled is disabled, Paimon sink will use original code.

Can you also re-run the failed action? It failed for some network issue when downloading JAR from maven repository. Re-run should fix this issue, but I do not have permission to trigger it.

cc @mikedias

[JingsongLi]

Spark's fast-write path still relies on the table-level coreOptions.bucket() setting rather than routing based on partition-to-bucket mappings; this can lead to errors or data being written to the incorrect bucket when partitions have varying bucket counts. The documentation itself acknowledges a risk of data corruption—an issue that cannot be mitigated simply by following usage guidelines. Another edge case arises in PartitionEntry.merge: when creation times are identical, the logic defaults to the larger bucket count, potentially causing the old layout to be retained during downscaling operations.