Skip to content

[flink] Support custom lookup shuffle for Flink lookup join. #3670

Description

@naivedogger

Search before asking

  • I searched in the issues and found nothing similar.

Motivation

Fluss's Flink lookup join doesn't custom-shuffle the probe stream, so probe rows are distributed to lookup subtasks arbitrarily. This hurts lookup.cache (partial) hit rate and enlarges RPC fan-out, since each subtask may hit any bucket / any TabletServer.
Flink 2.0 added SupportsLookupCustomShuffle (FLIP-462), which lets a source define the probe-stream distribution. We can shuffle by bucket key so each subtask handles a stable set of buckets → better cache locality and lower RPC fan-out.

Solution

  • Implement SupportsLookupCustomShuffle and provide an InputDataPartitioner that maps the lookup key to a Fluss bucket using the existing bucketing logic (consistent with the write path / PrimaryKeyLookuper).
  • Only enable when the lookup key covers the bucket key (PK lookup / prefix lookup on bucket key); otherwise fall back to default distribution.
  • Note: the interface is Flink 2.0+, so this targets fluss-flink-2.2 only (FlinkTableSource lives in fluss-flink-common which must stay Flink 1.18 compatible).

Anything else?

No response

Willingness to contribute

  • I'm willing to submit a PR!

Metadata

Metadata

Assignees

Labels

No labels
No labels

Fields

No fields configured for Feature.

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions