[flink] Support custom lookup shuffle by bucket for Flink 2.x

[naivedogger]

Purpose

Linked issue: close #3670

Implement SupportsLookupCustomShuffle (FLIP-462) in the Flink 2.2 connector so a
lookup-join probe stream is repartitioned by the Fluss bucket key. Rows that map to the
same Fluss bucket are co-located on the same lookup subtask, which improves lookup cache
locality and lowers RPC fan-out. The feature is opt-in through Flink's existing
LOOKUP('shuffle'='true') join hint; behaviour is unchanged when the hint is absent.

Brief change log

• fluss-flink-2.2: add FlinkLookupShuffleTableSource (extends FlinkTableSource,
  implements SupportsLookupCustomShuffle). getPartitioner() returns a bucket-based
  partitioner for full primary-key lookups; copy() preserves the subtype so the ability
  survives planner copies.
• fluss-flink-2.2: add FlussLookupInputPartitioner, which mirrors client-side
  bucketing (bucket-key encoding + BucketingFunction, selected by the table's data-lake
  format) and routes bucketId % numPartitions, matching how PrimaryKeyLookuper buckets.
• Cross-version isolation: add a same-FQN LookupShuffleSourceAdapter in both
  fluss-flink-common (no-op) and fluss-flink-2.2 (wraps the source). The 2.x class
  shadows the common one at package time, keeping fluss-flink-common compilable against
  Flink 1.20 (SupportsLookupCustomShuffle does not exist before Flink 2.0). This follows
  the existing *Adapter pattern in the module.
• fluss-flink-common FlinkTableSource: add a copy constructor as the single source of
  truth for state cloning (copy() now delegates to it); stash the LookupNormalizer
  computed in getLookupRuntimeProvider() for later use by getPartitioner(); expose a few
  protected accessors for the 2.x variant.
• fluss-flink-common FlinkTableFactory: route the created source through
  LookupShuffleSourceAdapter.maybeWithCustomShuffle(...).

Tests

• FlussLookupInputPartitionerTest (unit, 4 cases): partition() equals bucket % numPartitions,
  is deterministic, routes same-bucket keys to the same partition, and handles a composite
  bucket key — verified against an independent bucket oracle.
• Flink22LookupShuffleITCase (IT, 3 cases):
  • Non-partitioned and partitioned PK dim tables: inspect the StreamGraph and assert the
    lookup-join probe edge is partitioned by Flink's RowDataCustomStreamPartitioner (i.e.
    our partitioner was actually applied) only when the hint is present; plus result parity
    with/without the hint (custom shuffle is a pure distribution optimization).
  • testSameBucketKeysAreRoutedToSameSubtask: end-to-end via partitionCustom using the
    production partitioner — same bucket → same subtask, subtask == bucketId % parallelism,
    and keys spread across more than one subtask.
• mvn clean verify passes for the affected modules (fluss-flink-common, fluss-flink-2.2);
  spotless + checkstyle clean.

API and Format

No changes to public Fluss API or storage/wire format. This only adds a Flink 2.x
DynamicTableSource ability (SupportsLookupCustomShuffle) on the Flink connector side.

Documentation

No user-facing docs required; the feature is opt-in via the standard Flink
LOOKUP('shuffle'='true') hint. Current scope/limitations:

• Enabled for full primary-key lookups only; prefix lookups keep Flink's default distribution.
• Partitioned tables are shuffled by bucket only (no (partition, bucket) alignment).
• A temporary table created without an explicit bucket.num silently keeps the default
  distribution.

[zuston]

thanks for your work! @naivedogger . I have a minor question about this.

which improves lookup cache locality

Could you help describe more about this? I haven't seen the similar lookup cache related code.

[naivedogger]

Thanks @zuston for your comment!

For some context: I'm trying to help enhance Fluss's Flink connector, and this feature is similar to Paimon's Large Scale Lookup (Fixed Bucket) — it only kicks in on Flink 2.0+ and when the join keys contain all bucket keys.

What shuffle lookup buys us: rows of the same bucket are routed to the same lookup subtask, so each subtask only deals with its own buckets. This way, when partial caching is enabled, each subtask only needs to cache its own slice of the table, giving a potentially better cache hit ratio and less duplicated cached data across subtasks.

Here the cache refers to Flink's built-in lookup caching, not a Fluss-specific one — that's probably why it was hard to spot. FULL caching isn't supported yet; I'm planning to add it as a follow-up.

[zuston]

Thanks for your detailed explanation, @naivedogger. Much appreciated. 🙏