Skip to content

[flink] add support of hybrid lookup with lake #3887

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zuston wants to merge 6 commits into
base: main
Choose a base branch
from
Open

[flink] add support of hybrid lookup with lake #3887

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b545422
[flink] add support of hybrid lookup with lake
Aug 7, 2026
7f083b9
cache involved
Aug 7, 2026
8b98769
fix
Aug 7, 2026
8f08083
add better abstraction
Aug 7, 2026
3c47ec5
fix
Aug 7, 2026
0224645
add tests
Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions ...-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,33 @@ public class FlinkConnectorOptions {
+ "with the lookup key values. This feature cannot be used with PREFIX_LOOKUP type. "
+ "Default is false.");

public static final ConfigOption<Boolean> LOOKUP_LAKE_FALLBACK_ENABLED =
ConfigOptions.key("lookup.lake-fallback.enabled")
.booleanType()
.defaultValue(false)
.withDescription(
"Whether to fall back to the lake table when the Fluss partition is no longer available.");

public static final ConfigOption<Duration> LOOKUP_LAKE_FALLBACK_TIMEOUT =
ConfigOptions.key("lookup.lake-fallback.timeout")
.durationType()
.defaultValue(Duration.ofSeconds(30))
.withDescription("The timeout for a single lake fallback lookup.");

public static final ConfigOption<Integer> LOOKUP_LAKE_FALLBACK_EXECUTOR_THREADS =
ConfigOptions.key("lookup.lake-fallback.executor-threads")
.intType()
.defaultValue(4)
.withDescription(
"The number of worker threads used for blocking lake fallback lookups.");

public static final ConfigOption<Integer> LOOKUP_LAKE_FALLBACK_MAX_CONCURRENCY =
ConfigOptions.key("lookup.lake-fallback.max-concurrency")
.intType()
.defaultValue(1024)
.withDescription(
"The maximum number of active and queued lake fallback lookups per lookup function instance.");

// --------------------------------------------------------------------------------------------
// Scan specific options
// --------------------------------------------------------------------------------------------
Expand Down
8 changes: 8 additions & 0 deletions ...nk/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,10 @@ public DynamicTableSource createDynamicTableSource(Context context) {
startupOptions,
tableOptions.get(FlinkConnectorOptions.LOOKUP_ASYNC),
tableOptions.get(FlinkConnectorOptions.LOOKUP_INSERT_IF_NOT_EXISTS),
tableOptions.get(FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_ENABLED),
tableOptions.get(FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_TIMEOUT),
tableOptions.get(FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_EXECUTOR_THREADS),
tableOptions.get(FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_MAX_CONCURRENCY),
cache,
partitionDiscoveryIntervalMs,
splitAssignmentBatchSize,
Expand Down Expand Up @@ -242,6 +246,10 @@ public Set<ConfigOption<?>> optionalOptions() {
FlinkConnectorOptions.SCAN_KV_SNAPSHOT_LEASE_DURATION,
FlinkConnectorOptions.LOOKUP_ASYNC,
FlinkConnectorOptions.LOOKUP_INSERT_IF_NOT_EXISTS,
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_ENABLED,
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_TIMEOUT,
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_EXECUTOR_THREADS,
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_MAX_CONCURRENCY,
FlinkConnectorOptions.SINK_IGNORE_DELETE,
FlinkConnectorOptions.SINK_BUCKET_SHUFFLE,
FlinkConnectorOptions.SINK_DISTRIBUTION_MODE,
Expand Down
130 changes: 130 additions & 0 deletions ...link/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.fluss.flink.source.deserializer.RowDataDeserializationSchema;
import org.apache.fluss.flink.source.lookup.FlinkAsyncLookupFunction;
import org.apache.fluss.flink.source.lookup.FlinkLookupFunction;
import org.apache.fluss.flink.source.lookup.HybridLakeAsyncLookupFunction;
import org.apache.fluss.flink.source.lookup.LookupNormalizer;
import org.apache.fluss.flink.source.reader.LeaseContext;
import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils;
Expand All @@ -38,6 +39,7 @@
import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.lake.source.LakeSplit;
import org.apache.fluss.metadata.ChangelogImage;
import org.apache.fluss.metadata.DataLakeFormat;
import org.apache.fluss.metadata.DeleteBehavior;
import org.apache.fluss.metadata.MergeEngineType;
import org.apache.fluss.metadata.PartitionSpec;
Expand All @@ -57,6 +59,7 @@
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.ProviderContext;
import org.apache.flink.table.connector.RowLevelModificationScanContext;
Expand Down Expand Up @@ -90,6 +93,7 @@

import javax.annotation.Nullable;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
Expand Down Expand Up @@ -137,6 +141,10 @@ public class FlinkTableSource
// options for lookup source
private final boolean lookupAsync;
private final boolean insertIfNotExists;
private final boolean lakeFallbackEnabled;
private final Duration lakeFallbackTimeout;
private final int lakeFallbackExecutorThreads;
private final int lakeFallbackMaxConcurrency;
@Nullable private final LookupCache cache;

private final long scanPartitionDiscoveryIntervalMs;
Expand Down Expand Up @@ -208,6 +216,10 @@ public FlinkTableSource(
startupOptions,
lookupAsync,
insertIfNotExists,
false,
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_TIMEOUT.defaultValue(),
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_EXECUTOR_THREADS.defaultValue(),
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_MAX_CONCURRENCY.defaultValue(),
cache,
scanPartitionDiscoveryIntervalMs,
FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE.defaultValue(),
Expand Down Expand Up @@ -236,6 +248,54 @@ public FlinkTableSource(
@Nullable MergeEngineType mergeEngineType,
Map<String, String> tableOptions,
LeaseContext leaseContext) {
this(
tablePath,
flussConfig,
tableConfig,
tableOutputType,
primaryKeyIndexes,
bucketKeyIndexes,
partitionKeyIndexes,
streaming,
startupOptions,
lookupAsync,
insertIfNotExists,
false,
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_TIMEOUT.defaultValue(),
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_EXECUTOR_THREADS.defaultValue(),
FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_MAX_CONCURRENCY.defaultValue(),
cache,
scanPartitionDiscoveryIntervalMs,
splitPerAssignmentBatchSize,
isDataLakeEnabled,
mergeEngineType,
tableOptions,
leaseContext);
}

public FlinkTableSource(
TablePath tablePath,
Configuration flussConfig,
TableConfig tableConfig,
org.apache.flink.table.types.logical.RowType tableOutputType,
int[] primaryKeyIndexes,
int[] bucketKeyIndexes,
int[] partitionKeyIndexes,
boolean streaming,
FlinkConnectorOptionsUtils.StartupOptions startupOptions,
boolean lookupAsync,
boolean insertIfNotExists,
boolean lakeFallbackEnabled,
Duration lakeFallbackTimeout,
int lakeFallbackExecutorThreads,
int lakeFallbackMaxConcurrency,
@Nullable LookupCache cache,
long scanPartitionDiscoveryIntervalMs,
int splitPerAssignmentBatchSize,
boolean isDataLakeEnabled,
@Nullable MergeEngineType mergeEngineType,
Map<String, String> tableOptions,
LeaseContext leaseContext) {
this.tablePath = tablePath;
this.flussConfig = flussConfig;
this.tableOutputType = tableOutputType;
Expand All @@ -248,6 +308,10 @@ public FlinkTableSource(

this.lookupAsync = lookupAsync;
this.insertIfNotExists = insertIfNotExists;
this.lakeFallbackEnabled = lakeFallbackEnabled;
this.lakeFallbackTimeout = lakeFallbackTimeout;
this.lakeFallbackExecutorThreads = lakeFallbackExecutorThreads;
this.lakeFallbackMaxConcurrency = lakeFallbackMaxConcurrency;
this.cache = cache;

this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
Expand Down Expand Up @@ -477,6 +541,22 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
partitionKeyIndexes,
tableOutputType,
projectedFields);
if (lakeFallbackEnabled) {
validateLakeFallbackLookup(lookupNormalizer);
AsyncLookupFunction asyncLookupFunction =
new HybridLakeAsyncLookupFunction(
flussConfig,
tablePath,
tableOutputType,
primaryKeyIndexes,
lookupNormalizer,
projectedFields,
tableOptions,
lakeFallbackTimeout,
lakeFallbackExecutorThreads,
lakeFallbackMaxConcurrency);
return AsyncLookupFunctionProvider.of(asyncLookupFunction);
}
if (lookupAsync) {
AsyncLookupFunction asyncLookupFunction =
new FlinkAsyncLookupFunction(
Expand Down Expand Up @@ -508,6 +588,52 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
}
}

private void validateLakeFallbackLookup(LookupNormalizer lookupNormalizer) {
if (!lookupAsync) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' requires 'lookup.async' to be true.");
}
if (insertIfNotExists) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' cannot be used with 'lookup.insert-if-not-exists'.");
}
if (cache != null) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' cannot be used with lookup cache.");
}
if (!isDataLakeEnabled) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' requires a datalake-enabled Fluss table.");
}
if (!tableConfig.getDataLakeFormat().isPresent()
|| tableConfig.getDataLakeFormat().get() != DataLakeFormat.PAIMON) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' currently only supports Paimon lake tables.");
}
if (lookupNormalizer.getLookupType() != org.apache.fluss.client.lookup.LookupType.LOOKUP) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' only supports full primary-key lookup.");
}
if (partitionKeyIndexes.length == 0) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' requires a partitioned table.");
}
if (!tableConfig.getAutoPartitionStrategy().isAutoPartitionEnabled()) {
throw new TableException(
"Option 'lookup.lake-fallback.enabled' requires an auto-partitioned table.");
}
if (lakeFallbackExecutorThreads <= 0 || lakeFallbackMaxConcurrency <= 0) {
throw new TableException(
"Options 'lookup.lake-fallback.executor-threads' and "
+ "'lookup.lake-fallback.max-concurrency' must be positive.");
}
if (lakeFallbackExecutorThreads > lakeFallbackMaxConcurrency) {
throw new TableException(
"Option 'lookup.lake-fallback.executor-threads' must not exceed "
+ "'lookup.lake-fallback.max-concurrency'.");
}
}

@Override
public DynamicTableSource copy() {
FlinkTableSource source =
Expand All @@ -523,6 +649,10 @@ public DynamicTableSource copy() {
startupOptions,
lookupAsync,
insertIfNotExists,
lakeFallbackEnabled,
lakeFallbackTimeout,
lakeFallbackExecutorThreads,
lakeFallbackMaxConcurrency,
cache,
scanPartitionDiscoveryIntervalMs,
splitPerAssignmentBatchSize,
Expand Down
Loading
Loading