naivedogger opened a new pull request, #3675:
URL: https://github.com/apache/fluss/pull/3675
### 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
