[flink] size aware partitioner for full compaction job

[dwangatt]

Summary

This PR adds a new dedicated compaction bucket distribution strategy:

compaction.bucket-distribution-strategy=size-aware-batch

The new strategy is intended for bounded full compaction jobs. It improves compaction workload balance by assigning compact buckets based on their total input file size, instead of relying on the existing linear partition/bucket channel mapping.

Problem

The previous compaction distribution strategy could assign multiple heavy buckets to the same writer while other writers received much lighter work. This caused long-tail compaction tasks where a small number of writers determined the overall job runtime.
This was especially visible in high-parallelism full compaction jobs where the total workload was large, but writer-level input distribution was uneven.

What changed

When compaction.bucket-distribution-strategy=size-aware-batch is enabled for a batch full compaction job:

1. The compaction source enumerator groups input splits by (partition, bucket).
2. It calculates the total input data file size for each bucket group.
3. It uses size-aware bin packing to assign bucket groups to downstream writers.
4. All splits for the same (partition, bucket) are kept together and sent to the same writer.
5. The sink skips the old sink-side repartitioning step, preserving the source-side size-aware assignment.
6. The compaction writer starts a new operator chain to avoid chaining source and writer operators together.
   This keeps compaction correctness while improving writer-level workload balance.

Scope

The new strategy is only applied to:

batch full compaction
For streaming compaction or non-full compaction, the strategy falls back to the existing linear behavior.

Why bucket-level grouping is needed

A single (partition, bucket) can be split into multiple DataSplits. If size-aware assignment were done at the individual split level, splits from the same bucket could be sent to different writers, causing multiple writers to compact the same bucket and potentially conflict during commit.
To avoid this, the assignment is performed at the bucket-group level:

group key = partition + bucket
weight = sum(data file sizes across all splits in that bucket group)
This ensures that each bucket is compacted by only one writer while still balancing heavy buckets across the job.

Expected impact

This should reduce compaction long tail by distributing heavy buckets more evenly across writers. In benchmark runs, size-aware compaction significantly improved writer input balance and reduced overall full-compaction wall-clock time for large Paimon tables.

Tests

Added/updated tests for:

• parsing compaction.bucket-distribution-strategy
• ensuring size-aware-batch is only used for batch full compaction
• source parallelism selection for size-aware compaction
• bucket file-size weight calculation
• grouping splits by (partition, bucket)
• ensuring splits from the same bucket group are assigned to the same downstream writer
• preserving existing fair split assignment behavior when no grouping function is provided

We also verified this change in compaction job. The new stragety can reduce compaction time from 5 hours to 1.5 hours without any significant long tail workers processing heavy buckets.

[JingsongLi]

When size-aware-batch is enabled and scan.split-enumerator.mode=preemptive, the source ignores the bucket grouping function, but the sink skips the bucket shuffle. Multiple splits from the same bucket may end up in different writers, resulting in concurrent full compaction. It is recommended to enforce or validate FAIR, or to implement grouping and allocation for PREEMPTIVE.