[flink] add support of hybrid lookup with lake - #3887
Open
zuston wants to merge 6 commits into
Open
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
Adds Paimon-backed hybrid async lookups for auto-partitioned primary-key tables, falling back to lake storage when a Fluss partition is unavailable.
Changes:
- Adds configurable lake fallback and runtime validation.
- Introduces shared Fluss/lake lookup runtimes and result conversion.
- Adds factory and integration tests with a test Paimon plugin.
Reviewed changes
Copilot reviewed 14 out of 15 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
.../LakeStoragePlugin |
Registers the test Paimon plugin. |
.../log4j2-test.properties |
No semantic configuration change. |
.../TestingPaimonLakeStoragePlugin.java |
Implements test lake lookup behavior. |
.../HybridLakeAsyncLookupFunctionTest.java |
Tests fallback, errors, and timeout. |
.../FlinkTableFactoryTest.java |
Tests provider selection and validation. |
.../LookupRuntime.java |
Defines the shared runtime interface. |
.../LookupResultConverter.java |
Centralizes projection and conversion. |
.../LakeLookupRuntime.java |
Executes Paimon lake lookups. |
.../HybridLakeAsyncLookupFunction.java |
Coordinates Fluss and lake lookup paths. |
.../FlussLookupRuntime.java |
Encapsulates Fluss lookup resources. |
.../FlinkLookupFunction.java |
Uses the shared synchronous runtime. |
.../FlinkAsyncLookupFunction.java |
Uses the shared asynchronous runtime. |
.../FlinkTableSource.java |
Selects and validates hybrid lookup. |
.../FlinkConnectorOptions.java |
Defines lake fallback options. |
.../FlinkTableFactory.java |
Wires options into table sources. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+40
to
+58
| private final int[] projection; | ||
| private transient volatile FlussRowToFlinkRowConverter rowConverter; | ||
|
|
||
| LookupResultConverter(RowType outputRowType, int[] projection) { | ||
| this.projection = projection; | ||
| this.rowConverter = new FlussRowToFlinkRowConverter(outputRowType); | ||
| } | ||
|
|
||
| Collection<RowData> convert( | ||
| @Nullable List<InternalRow> lookupRows, | ||
| @Nullable LookupNormalizer.RemainingFilter remainingFilter) { | ||
| if (lookupRows == null || lookupRows.isEmpty()) { | ||
| return Collections.emptyList(); | ||
| } | ||
|
|
||
| List<RowData> projectedRows = new ArrayList<>(lookupRows.size()); | ||
| for (InternalRow row : lookupRows) { | ||
| if (row != null) { | ||
| RowData flinkRow = rowConverter.toFlinkRowData(maybeProject(row)); |
Comment on lines
+158
to
+160
| partitionExists = !admin.listPartitionInfos(tablePath, partitionSpec).get().isEmpty(); | ||
| partitionExistenceCache.put(partitionSpec, partitionExists); | ||
| return partitionExists; |
| } | ||
| } | ||
|
|
||
| private synchronized boolean getOrRefreshPartitionExistence(PartitionSpec partitionSpec) { |
Comment on lines
+153
to
+156
| executor.execute( | ||
| () -> { | ||
| try { | ||
| byte[] value = lookuper.lookup(keyBytes, lookupContext); |
Member
|
I think #3630 already can cover this feature... |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
When Fluss is used as the hot layer and the lake as the cold layer for primary-key tables, the Flink connector previously had no way to look up data across both layers. This PR adds a hybrid lookup mechanism for auto-partitioned tables that determines whether a lookup key belongs to a live Fluss partition or should fall back to the lake.
Brief change log
When
lookup.lake-fallback.enabled=true, lookup tries Fluss first and falls back to an async lake point lookup if the partition is missing. This is built on a sharedLookupRuntimeabstraction (FlussLookupRuntime+LakeLookupRuntime) and a newHybridLakeAsyncLookupFunction.Currently supports Paimon, async lookup, auto-partitioned tables, and full primary-key lookup only.
Tests
API and Format
Documentation