From b5454222ad3c3859c5a3ca3f743c18fe8b681124 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Fri, 7 Aug 2026 11:46:35 +0800 Subject: [PATCH 1/6] [flink] add support of hybrid lookup with lake --- .../fluss/flink/FlinkConnectorOptions.java | 27 ++ .../flink/catalog/FlinkTableFactory.java | 8 + .../fluss/flink/source/FlinkTableSource.java | 127 +++++++ .../lookup/FlinkAsyncLookupFunction.java | 178 +++------ .../source/lookup/FlinkLookupFunction.java | 110 ++---- .../source/lookup/FlussLookupRuntime.java | 113 ++++++ .../lookup/HybridLakeAsyncLookupFunction.java | 358 ++++++++++++++++++ .../source/lookup/LakeLookupRuntime.java | 208 ++++++++++ .../source/lookup/LookupResultConverter.java | 65 ++++ .../flink/catalog/FlinkTableFactoryTest.java | 129 ++++++- .../TestingPaimonLakeStoragePlugin.java | 111 ++++++ ...e.fluss.lake.lakestorage.LakeStoragePlugin | 3 +- 12 files changed, 1230 insertions(+), 207 deletions(-) create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java create mode 100644 fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java index 23d8f0b2e9..00fcba1905 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java @@ -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 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 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 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 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 // -------------------------------------------------------------------------------------------- diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java index ab64959abd..111b5fe5b6 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java @@ -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, @@ -242,6 +246,10 @@ public Set> 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, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java index 782c8c49f2..7a8653d7fd 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java @@ -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; @@ -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; @@ -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; @@ -90,6 +93,7 @@ import javax.annotation.Nullable; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -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; @@ -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(), @@ -236,6 +248,54 @@ public FlinkTableSource( @Nullable MergeEngineType mergeEngineType, Map 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 tableOptions, + LeaseContext leaseContext) { this.tablePath = tablePath; this.flussConfig = flussConfig; this.tableOutputType = tableOutputType; @@ -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; @@ -477,6 +541,23 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { partitionKeyIndexes, tableOutputType, projectedFields); + if (lakeFallbackEnabled) { + validateLakeFallbackLookup(lookupNormalizer); + AsyncLookupFunction asyncLookupFunction = + new HybridLakeAsyncLookupFunction( + flussConfig, + tablePath, + tableOutputType, + primaryKeyIndexes, + partitionKeyIndexes, + lookupNormalizer, + projectedFields, + tableOptions, + lakeFallbackTimeout, + lakeFallbackExecutorThreads, + lakeFallbackMaxConcurrency); + return AsyncLookupFunctionProvider.of(asyncLookupFunction); + } if (lookupAsync) { AsyncLookupFunction asyncLookupFunction = new FlinkAsyncLookupFunction( @@ -508,6 +589,48 @@ 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 (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 = @@ -523,6 +646,10 @@ public DynamicTableSource copy() { startupOptions, lookupAsync, insertIfNotExists, + lakeFallbackEnabled, + lakeFallbackTimeout, + lakeFallbackExecutorThreads, + lakeFallbackMaxConcurrency, cache, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java index 78b4fb8ed9..437ac72fe9 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java @@ -17,22 +17,11 @@ package org.apache.fluss.flink.source.lookup; -import org.apache.fluss.client.Connection; -import org.apache.fluss.client.ConnectionFactory; -import org.apache.fluss.client.lookup.Lookup; -import org.apache.fluss.client.lookup.LookupType; -import org.apache.fluss.client.lookup.Lookuper; -import org.apache.fluss.client.table.Table; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.TableNotExistException; -import org.apache.fluss.flink.row.FlinkAsFlussRow; -import org.apache.fluss.flink.source.lookup.LookupNormalizer.RemainingFilter; import org.apache.fluss.flink.utils.FlinkConversions; import org.apache.fluss.flink.utils.FlinkUtils; -import org.apache.fluss.flink.utils.FlussRowToFlinkRowConverter; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.row.InternalRow; -import org.apache.fluss.row.ProjectedRow; import org.apache.fluss.utils.ExceptionUtils; import org.apache.flink.table.data.RowData; @@ -44,13 +33,12 @@ import javax.annotation.Nullable; -import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; -import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.stream.IntStream; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + /** A flink async lookup function for fluss. */ public class FlinkAsyncLookupFunction extends AsyncLookupFunction { @@ -58,18 +46,12 @@ public class FlinkAsyncLookupFunction extends AsyncLookupFunction { private static final long serialVersionUID = 1L; - private final Configuration flussConfig; private final TablePath tablePath; private final RowType flinkRowType; private final LookupNormalizer lookupNormalizer; - @Nullable private int[] projection; - private final boolean insertIfNotExists; - - private transient FlussRowToFlinkRowConverter flussRowToFlinkRowConverter; - private transient Connection connection; - private transient Table table; - private transient Lookuper lookuper; - private transient FlinkAsFlussRow lookupRow; + private final int[] projection; + private final FlussLookupRuntime flussLookupRuntime; + private transient LookupResultConverter lookupResultConverter; public FlinkAsyncLookupFunction( Configuration flussConfig, @@ -78,46 +60,26 @@ public FlinkAsyncLookupFunction( LookupNormalizer lookupNormalizer, @Nullable int[] projection, boolean insertIfNotExists) { - this.flussConfig = flussConfig; this.tablePath = tablePath; this.flinkRowType = flinkRowType; this.lookupNormalizer = lookupNormalizer; - this.projection = projection; - this.insertIfNotExists = insertIfNotExists; + this.projection = + projection == null + ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() + : projection; + this.flussLookupRuntime = + new FlussLookupRuntime( + flussConfig, tablePath, flinkRowType, lookupNormalizer, insertIfNotExists); } @Override public void open(FunctionContext context) { LOG.info("start open ..."); - connection = ConnectionFactory.createConnection(flussConfig); - table = connection.getTable(tablePath); - lookupRow = new FlinkAsFlussRow(); - - final RowType outputRowType; - if (projection == null) { - outputRowType = flinkRowType; - // we force to do projection if no projection pushdown, in order to handle schema - // changes (ADD COLUMN LAST), this guarantees the input row of - // flussRowToFlinkRowConverter is in expected schema even new columns are added. - projection = IntStream.range(0, flinkRowType.getFieldCount()).toArray(); - } else { - outputRowType = FlinkUtils.projectRowType(flinkRowType, projection); - } - // TODO: currently, we assume only ADD COLUMN LAST schema changes, so the projection - // positions can still work even after such changes. - flussRowToFlinkRowConverter = - new FlussRowToFlinkRowConverter(FlinkConversions.toFlussRowType(outputRowType)); - - Lookup lookup = table.newLookup(); - if (lookupNormalizer.getLookupType() == LookupType.PREFIX_LOOKUP) { - int[] lookupKeyIndexes = lookupNormalizer.getLookupKeyIndexes(); - RowType lookupKeyRowType = FlinkUtils.projectRowType(flinkRowType, lookupKeyIndexes); - lookup = lookup.lookupBy(lookupKeyRowType.getFieldNames()); - } else if (insertIfNotExists) { - lookup = lookup.enableInsertIfNotExists(); - } - lookuper = lookup.createLookuper(); - + flussLookupRuntime.open(); + RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, projection); + lookupResultConverter = + new LookupResultConverter( + FlinkConversions.toFlussRowType(outputRowType), projection); LOG.info("end open."); } @@ -129,77 +91,63 @@ public void open(FunctionContext context) { @Override public CompletableFuture> asyncLookup(RowData keyRow) { RowData normalizedKeyRow = lookupNormalizer.normalizeLookupKey(keyRow); - RemainingFilter remainingFilter = lookupNormalizer.createRemainingFilter(keyRow); - InternalRow flussKeyRow = lookupRow.replace(normalizedKeyRow); + LookupNormalizer.RemainingFilter remainingFilter = + lookupNormalizer.createRemainingFilter(keyRow); // the retry mechanism is now handled by the underlying LookupClient layer, // we can't call lookuper.lookup() in whenComplete callback as lookuper is not thread-safe. CompletableFuture> future = new CompletableFuture<>(); - lookuper.lookup(flussKeyRow) - .whenComplete( - (result, throwable) -> { - if (throwable != null) { - if (ExceptionUtils.findThrowable( - throwable, TableNotExistException.class) - .isPresent()) { - LOG.error("Table '{}' not found ", tablePath, throwable); - future.completeExceptionally( - new RuntimeException( - "Fluss table '" + tablePath + "' not found.", - throwable)); - } else { - LOG.error("Fluss asyncLookup error", throwable); - future.completeExceptionally( - new RuntimeException( - "Execution of Fluss asyncLookup failed: " - + throwable.getMessage(), - throwable)); + try { + flussLookupRuntime + .lookup(normalizedKeyRow) + .whenComplete( + (result, throwable) -> { + try { + if (throwable != null) { + if (ExceptionUtils.findThrowable( + throwable, TableNotExistException.class) + .isPresent()) { + LOG.error( + "Table '{}' not found ", tablePath, throwable); + future.completeExceptionally( + new RuntimeException( + "Fluss table '" + + tablePath + + "' not found.", + throwable)); + } else { + LOG.error("Fluss asyncLookup error", throwable); + future.completeExceptionally( + new RuntimeException( + "Execution of Fluss asyncLookup failed: " + + throwable.getMessage(), + throwable)); + } + } else { + future.complete( + checkNotNull( + lookupResultConverter, + "Lookup result converter is not initialized.") + .convert( + result == null + ? null + : result.getRowList(), + remainingFilter)); + } + } catch (Throwable t) { + future.completeExceptionally(t); } - } else { - handleLookupSuccess(future, result.getRowList(), remainingFilter); - } - }); - return future; - } - - private void handleLookupSuccess( - CompletableFuture> resultFuture, - List lookupResult, - @Nullable RemainingFilter remainingFilter) { - if (lookupResult.isEmpty()) { - resultFuture.complete(Collections.emptyList()); - return; - } - - List projectedRow = new ArrayList<>(); - for (InternalRow row : lookupResult) { - if (row != null) { - RowData flinkRow = flussRowToFlinkRowConverter.toFlinkRowData(maybeProject(row)); - if (remainingFilter == null || remainingFilter.isMatch(flinkRow)) { - projectedRow.add(flinkRow); - } - } + }); + } catch (Throwable t) { + future.completeExceptionally(t); } - resultFuture.complete(projectedRow); - } - - private InternalRow maybeProject(InternalRow row) { - if (projection == null) { - return row; - } - // should not reuse objects for async operations - return ProjectedRow.from(projection).replaceRow(row); + return future; } @Override public void close() throws Exception { LOG.info("start close ..."); - if (table != null) { - table.close(); - } - if (connection != null) { - connection.close(); - } + flussLookupRuntime.close(); LOG.info("end close."); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java index a2bda2feca..e9b051a84e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java @@ -17,20 +17,11 @@ package org.apache.fluss.flink.source.lookup; -import org.apache.fluss.client.Connection; -import org.apache.fluss.client.ConnectionFactory; -import org.apache.fluss.client.lookup.Lookup; -import org.apache.fluss.client.lookup.LookupType; -import org.apache.fluss.client.lookup.Lookuper; -import org.apache.fluss.client.table.Table; import org.apache.fluss.config.Configuration; -import org.apache.fluss.flink.row.FlinkAsFlussRow; import org.apache.fluss.flink.utils.FlinkConversions; import org.apache.fluss.flink.utils.FlinkUtils; -import org.apache.fluss.flink.utils.FlussRowToFlinkRowConverter; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.row.InternalRow; -import org.apache.fluss.row.ProjectedRow; import org.apache.flink.table.data.RowData; import org.apache.flink.table.functions.FunctionContext; @@ -41,31 +32,23 @@ import javax.annotation.Nullable; -import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.List; import java.util.stream.IntStream; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + /** A flink lookup function for fluss. */ public class FlinkLookupFunction extends LookupFunction { private static final Logger LOG = LoggerFactory.getLogger(FlinkLookupFunction.class); private static final long serialVersionUID = 1L; - private final Configuration flussConfig; - private final TablePath tablePath; private final RowType flinkRowType; private final LookupNormalizer lookupNormalizer; - @Nullable private final int[] projection; - private final boolean insertIfNotExists; - - private transient FlussRowToFlinkRowConverter flussRowToFlinkRowConverter; - private transient Connection connection; - private transient Table table; - private transient Lookuper lookuper; - private transient FlinkAsFlussRow lookupRow; - @Nullable private transient ProjectedRow projectedRow; + private final int[] projection; + private final FlussLookupRuntime flussLookupRuntime; + private transient LookupResultConverter lookupResultConverter; public FlinkLookupFunction( Configuration flussConfig, @@ -74,49 +57,25 @@ public FlinkLookupFunction( LookupNormalizer lookupNormalizer, @Nullable int[] projection, boolean insertIfNotExists) { - this.flussConfig = flussConfig; - this.tablePath = tablePath; this.flinkRowType = flinkRowType; this.lookupNormalizer = lookupNormalizer; - this.projection = projection; - this.insertIfNotExists = insertIfNotExists; + this.projection = + projection == null + ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() + : projection; + this.flussLookupRuntime = + new FlussLookupRuntime( + flussConfig, tablePath, flinkRowType, lookupNormalizer, insertIfNotExists); } @Override public void open(FunctionContext context) { LOG.info("start open ..."); - connection = ConnectionFactory.createConnection(flussConfig); - table = connection.getTable(tablePath); - lookupRow = new FlinkAsFlussRow(); - - final RowType outputRowType; - if (projection == null) { - outputRowType = flinkRowType; - // we force to do projection if no projection pushdown, in order to handle schema - // changes (ADD COLUMN LAST), this guarantees the input row of - // flussRowToFlinkRowConverter is in expected schema even new columns are added. - projectedRow = - ProjectedRow.from(IntStream.range(0, flinkRowType.getFieldCount()).toArray()); - } else { - outputRowType = FlinkUtils.projectRowType(flinkRowType, projection); - // reuse the projected row - projectedRow = ProjectedRow.from(projection); - } - // TODO: currently, we assume only ADD COLUMN LAST schema changes, so the projection - // positions can still work even after such changes. - flussRowToFlinkRowConverter = - new FlussRowToFlinkRowConverter(FlinkConversions.toFlussRowType(outputRowType)); - - Lookup lookup = table.newLookup(); - if (lookupNormalizer.getLookupType() == LookupType.PREFIX_LOOKUP) { - int[] lookupKeyIndexes = lookupNormalizer.getLookupKeyIndexes(); - RowType lookupKeyRowType = FlinkUtils.projectRowType(flinkRowType, lookupKeyIndexes); - lookup = lookup.lookupBy(lookupKeyRowType.getFieldNames()); - } else if (insertIfNotExists) { - lookup = lookup.enableInsertIfNotExists(); - } - lookuper = lookup.createLookuper(); - + flussLookupRuntime.open(); + RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, projection); + lookupResultConverter = + new LookupResultConverter( + FlinkConversions.toFlussRowType(outputRowType), projection); LOG.info("end open."); } @@ -131,48 +90,23 @@ public Collection lookup(RowData keyRow) { RowData normalizedKeyRow = lookupNormalizer.normalizeLookupKey(keyRow); LookupNormalizer.RemainingFilter remainingFilter = lookupNormalizer.createRemainingFilter(keyRow); - // wrap flink row as fluss row to lookup, the flink row has already been in expected order. - InternalRow flussKeyRow = lookupRow.replace(normalizedKeyRow); // the retry mechanism will be handled by the underlying LookupClient layer try { - List lookupRows = lookuper.lookup(flussKeyRow).get().getRowList(); - if (lookupRows.isEmpty()) { - return Collections.emptyList(); - } - List projectedRows = new ArrayList<>(); - for (InternalRow row : lookupRows) { - if (row != null) { - RowData flinkRow = - flussRowToFlinkRowConverter.toFlinkRowData(maybeProject(row)); - if (remainingFilter == null || remainingFilter.isMatch(flinkRow)) { - projectedRows.add(flinkRow); - } - } - } - return projectedRows; + List rows = flussLookupRuntime.lookup(normalizedKeyRow).get().getRowList(); + return checkNotNull( + lookupResultConverter, "Lookup result converter is not initialized.") + .convert(rows, remainingFilter); } catch (Exception e) { LOG.error("Fluss lookup error", e); throw new RuntimeException("Execution of Fluss lookup failed: " + e.getMessage(), e); } } - private InternalRow maybeProject(InternalRow row) { - if (projectedRow == null) { - return row; - } - return projectedRow.replaceRow(row); - } - @Override public void close() throws Exception { LOG.info("start close ..."); - if (table != null) { - table.close(); - } - if (connection != null) { - connection.close(); - } + flussLookupRuntime.close(); LOG.info("end close."); } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java new file mode 100644 index 0000000000..1e363aa0ca --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java @@ -0,0 +1,113 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.source.lookup; + +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.lookup.Lookup; +import org.apache.fluss.client.lookup.LookupResult; +import org.apache.fluss.client.lookup.LookupType; +import org.apache.fluss.client.lookup.Lookuper; +import org.apache.fluss.client.table.Table; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.row.FlinkAsFlussRow; +import org.apache.fluss.flink.utils.FlinkUtils; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.logical.RowType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Serializable; +import java.util.concurrent.CompletableFuture; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** The shared Fluss client-side lookup runtime. */ +final class FlussLookupRuntime implements Serializable { + + private static final Logger LOG = LoggerFactory.getLogger(FlussLookupRuntime.class); + + private static final long serialVersionUID = 1L; + + private final Configuration flussConfig; + private final TablePath tablePath; + private final RowType flinkRowType; + private final LookupNormalizer lookupNormalizer; + private final boolean insertIfNotExists; + + private transient Connection connection; + private transient Table table; + private transient Lookuper lookuper; + + FlussLookupRuntime( + Configuration flussConfig, + TablePath tablePath, + RowType flinkRowType, + LookupNormalizer lookupNormalizer, + boolean insertIfNotExists) { + this.flussConfig = flussConfig; + this.tablePath = tablePath; + this.flinkRowType = flinkRowType; + this.lookupNormalizer = lookupNormalizer; + this.insertIfNotExists = insertIfNotExists; + } + + void open() { + LOG.info("Starting Fluss lookup runtime for table {}.", tablePath); + connection = ConnectionFactory.createConnection(flussConfig); + table = connection.getTable(tablePath); + + Lookup lookup = table.newLookup(); + if (lookupNormalizer.getLookupType() == LookupType.PREFIX_LOOKUP) { + int[] lookupKeyIndexes = lookupNormalizer.getLookupKeyIndexes(); + RowType lookupKeyRowType = FlinkUtils.projectRowType(flinkRowType, lookupKeyIndexes); + lookup = lookup.lookupBy(lookupKeyRowType.getFieldNames()); + } else if (insertIfNotExists) { + lookup = lookup.enableInsertIfNotExists(); + } + lookuper = lookup.createLookuper(); + LOG.info("Finished starting Fluss lookup runtime."); + } + + CompletableFuture lookup(RowData normalizedKeyRow) { + return checkNotNull(lookuper, "Fluss lookuper must be initialized.") + .lookup(new FlinkAsFlussRow(normalizedKeyRow)); + } + + TableInfo getTableInfo() { + return checkNotNull(table, "Fluss table must be initialized.").getTableInfo(); + } + + Admin getAdmin() { + return checkNotNull(connection, "Fluss connection must be initialized.").getAdmin(); + } + + void close() throws Exception { + LOG.info("Closing Fluss lookup runtime for table {}.", tablePath); + if (table != null) { + table.close(); + } + if (connection != null) { + connection.close(); + } + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java new file mode 100644 index 0000000000..06815903cf --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java @@ -0,0 +1,358 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.source.lookup; + +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.lookup.LookupType; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.PartitionNotExistException; +import org.apache.fluss.flink.utils.FlinkConversions; +import org.apache.fluss.flink.utils.FlinkUtils; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.utils.ExceptionUtils; +import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; + +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.functions.AsyncLookupFunction; +import org.apache.flink.table.functions.FunctionContext; +import org.apache.flink.table.types.logical.RowType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.stream.IntStream; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * An async lookup function that first looks up Fluss and falls back to a lake point lookup when the + * requested Fluss partition is absent. + * + *

The lake fallback currently supports Paimon tables only. It is deliberately a point lookup: no + * lake source or lake split planner is involved. + */ +public class HybridLakeAsyncLookupFunction extends AsyncLookupFunction { + + private static final Logger LOG = LoggerFactory.getLogger(HybridLakeAsyncLookupFunction.class); + private static final long serialVersionUID = 1L; + + private final TablePath tablePath; + private final LookupNormalizer lookupNormalizer; + private final FlussLookupRuntime flussLookupRuntime; + private transient LookupResultConverter lookupResultConverter; + + private final LakeLookupRuntime lakeLookupRuntime; + private final Duration lakeFallbackTimeout; + private final int lakeFallbackExecutorThreads; + private final int lakeFallbackMaxConcurrency; + private transient ThreadPoolExecutor lakeLookupExecutor; + private transient ScheduledExecutorService timeoutExecutor; + + public HybridLakeAsyncLookupFunction( + Configuration flussConfig, + TablePath tablePath, + RowType flinkRowType, + int[] primaryKeyIndexes, + int[] partitionKeyIndexes, + LookupNormalizer lookupNormalizer, + @Nullable int[] projection, + Map tableOptions, + Duration lakeFallbackTimeout, + int lakeFallbackExecutorThreads, + int lakeFallbackMaxConcurrency) { + this.tablePath = tablePath; + this.lookupNormalizer = lookupNormalizer; + + this.lakeFallbackTimeout = lakeFallbackTimeout; + this.lakeFallbackExecutorThreads = lakeFallbackExecutorThreads; + this.lakeFallbackMaxConcurrency = lakeFallbackMaxConcurrency; + + validateLookupShape( + primaryKeyIndexes, + partitionKeyIndexes, + lookupNormalizer, + lakeFallbackTimeout, + lakeFallbackExecutorThreads, + lakeFallbackMaxConcurrency); + + int[] resolvedProjection = + projection == null + ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() + : projection; + this.lookupResultConverter = + new LookupResultConverter( + FlinkConversions.toFlussRowType( + FlinkUtils.projectRowType(flinkRowType, resolvedProjection)), + resolvedProjection); + + this.flussLookupRuntime = + new FlussLookupRuntime( + flussConfig, tablePath, flinkRowType, lookupNormalizer, false); + this.lakeLookupRuntime = + new LakeLookupRuntime( + flussConfig, + tablePath, + FlinkConversions.toFlussRowType(flinkRowType), + primaryKeyIndexes, + tableOptions); + } + + @Override + public void open(@Nullable FunctionContext context) { + LOG.info("Starting hybrid lake async lookup function for table {}.", tablePath); + flussLookupRuntime.open(); + + TableInfo tableInfo = flussLookupRuntime.getTableInfo(); + lakeLookupRuntime.open(tableInfo); + + lakeLookupExecutor = createLakeLookupExecutor(); + timeoutExecutor = + new ScheduledThreadPoolExecutor( + 1, new ExecutorThreadFactory("fluss-lake-fallback-timeout")); + LOG.info("Finished opening hybrid lake async lookup function for table {}.", tablePath); + } + + @Override + public CompletableFuture> asyncLookup(RowData keyRow) { + RowData normalizedKeyRow = lookupNormalizer.normalizeLookupKey(keyRow); + LookupNormalizer.RemainingFilter remainingFilter = + lookupNormalizer.createRemainingFilter(keyRow); + LakeLookupRuntime.LakeLookupKey lakeLookupKey = + lakeLookupRuntime.createLookupKey(normalizedKeyRow); + + CompletableFuture> future = new CompletableFuture<>(); + try { + flussLookupRuntime + .lookup(normalizedKeyRow) + .whenComplete( + (result, throwable) -> { + try { + if (throwable != null) { + if (ExceptionUtils.findThrowable( + throwable, PartitionNotExistException.class) + .isPresent()) { + checkPartitionAndLookupLake( + lakeLookupKey, remainingFilter, future); + return; + } + LOG.error( + "Fluss async lookup failed for table {}.", + tablePath, + throwable); + future.completeExceptionally( + new RuntimeException( + "Execution of Fluss async lookup failed: " + + throwable.getMessage(), + throwable)); + return; + } + + boolean hit = result != null && !result.getRowList().isEmpty(); + if (hit) { + future.complete( + checkNotNull( + lookupResultConverter, + "Lookup result converter is not initialized.") + .convert( + result.getRowList(), + remainingFilter)); + } else { + checkPartitionAndLookupLake( + lakeLookupKey, remainingFilter, future); + } + } catch (Throwable t) { + future.completeExceptionally(t); + } + }); + } catch (Throwable t) { + if (ExceptionUtils.findThrowable(t, PartitionNotExistException.class).isPresent()) { + checkPartitionAndLookupLake(lakeLookupKey, remainingFilter, future); + } else { + future.completeExceptionally(t); + } + } + return future; + } + + private void checkPartitionAndLookupLake( + LakeLookupRuntime.LakeLookupKey lakeLookupKey, + @Nullable LookupNormalizer.RemainingFilter remainingFilter, + CompletableFuture> future) { + Admin admin = flussLookupRuntime.getAdmin(); + admin.listPartitionInfos(tablePath, lakeLookupKey.getPartitionSpec().toPartitionSpec()) + .whenComplete( + (partitionInfos, throwable) -> { + try { + if (throwable != null) { + future.completeExceptionally(throwable); + return; + } + boolean partitionExists = !partitionInfos.isEmpty(); + if (partitionExists) { + future.complete(Collections.emptyList()); + } else { + lookupLakeAsync(lakeLookupKey, remainingFilter, future); + } + } catch (Throwable t) { + future.completeExceptionally(t); + } + }); + } + + private ThreadPoolExecutor createLakeLookupExecutor() { + int queueCapacity = lakeFallbackMaxConcurrency - lakeFallbackExecutorThreads; + BlockingQueue queue = + queueCapacity == 0 + ? new SynchronousQueue<>() + : new ArrayBlockingQueue<>(queueCapacity); + return new ThreadPoolExecutor( + lakeFallbackExecutorThreads, + lakeFallbackExecutorThreads, + 0L, + TimeUnit.MILLISECONDS, + queue, + new ExecutorThreadFactory("fluss-lake-fallback-lookup"), + new ThreadPoolExecutor.AbortPolicy()); + } + + private void lookupLakeAsync( + LakeLookupRuntime.LakeLookupKey lakeLookupKey, + @Nullable LookupNormalizer.RemainingFilter remainingFilter, + CompletableFuture> future) { + ScheduledFuture timeoutTask; + try { + timeoutTask = + timeoutExecutor.schedule( + () -> + completeLakeFallbackExceptionally( + future, + new TimeoutException( + "Lake fallback lookup timed out after " + + lakeFallbackTimeout)), + lakeFallbackTimeout.toMillis(), + TimeUnit.MILLISECONDS); + future.whenComplete((ignored, ignoredError) -> timeoutTask.cancel(false)); + } catch (RejectedExecutionException e) { + completeLakeFallbackExceptionally( + future, new RuntimeException("Lake fallback timeout executor is closed.", e)); + return; + } + + try { + lakeLookupExecutor.execute( + () -> { + try { + Collection rows = lookupLake(lakeLookupKey, remainingFilter); + completeLakeFallbackSuccessfully(future, rows); + } catch (Throwable t) { + completeLakeFallbackExceptionally( + future, + new RuntimeException( + "Execution of lake fallback lookup failed: " + + t.getMessage(), + t)); + } + }); + } catch (RejectedExecutionException e) { + completeLakeFallbackExceptionally( + future, + new RuntimeException("Lake fallback lookup executor is overloaded.", e)); + } + } + + private void completeLakeFallbackSuccessfully( + CompletableFuture> future, Collection rows) { + future.complete(rows); + } + + private void completeLakeFallbackExceptionally( + CompletableFuture> future, Throwable throwable) { + future.completeExceptionally(throwable); + } + + private Collection lookupLake( + LakeLookupRuntime.LakeLookupKey lakeLookupKey, + @Nullable LookupNormalizer.RemainingFilter remainingFilter) + throws Exception { + InternalRow row = lakeLookupRuntime.lookup(lakeLookupKey); + if (row == null) { + return Collections.emptyList(); + } + return lookupResultConverter.convert(Collections.singletonList(row), remainingFilter); + } + + private static void validateLookupShape( + int[] primaryKeyIndexes, + int[] partitionKeyIndexes, + LookupNormalizer lookupNormalizer, + Duration lakeFallbackTimeout, + int lakeFallbackExecutorThreads, + int lakeFallbackMaxConcurrency) { + if (primaryKeyIndexes.length == 0) { + throw new TableException("Lake fallback lookup requires a primary-key table."); + } + if (partitionKeyIndexes.length == 0) { + throw new TableException("Lake fallback lookup requires a partitioned table."); + } + if (lookupNormalizer.getLookupType() != LookupType.LOOKUP) { + throw new TableException("Lake fallback lookup only supports full primary-key lookup."); + } + if (lakeFallbackTimeout.isZero() || lakeFallbackTimeout.isNegative()) { + throw new TableException("Lake fallback lookup timeout must be positive."); + } + if (lakeFallbackExecutorThreads <= 0 || lakeFallbackMaxConcurrency <= 0) { + throw new TableException("Lake fallback lookup executor settings must be positive."); + } + if (lakeFallbackExecutorThreads > lakeFallbackMaxConcurrency) { + throw new TableException( + "Lake fallback lookup executor threads must not exceed max concurrency."); + } + } + + @Override + public void close() throws Exception { + LOG.info("Closing hybrid lake async lookup function for table {}.", tablePath); + if (lakeLookupExecutor != null) { + lakeLookupExecutor.shutdownNow(); + } + if (timeoutExecutor != null) { + timeoutExecutor.shutdownNow(); + } + lakeLookupRuntime.close(); + flussLookupRuntime.close(); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java new file mode 100644 index 0000000000..283c2f158a --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java @@ -0,0 +1,208 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.source.lookup; + +import org.apache.fluss.bucketing.BucketingFunction; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.row.FlinkAsFlussRow; +import org.apache.fluss.flink.utils.DataLakeUtils; +import org.apache.fluss.lake.lakestorage.LakeStorage; +import org.apache.fluss.lake.lakestorage.LakeStoragePlugin; +import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp; +import org.apache.fluss.lake.lakestorage.LakeTableLookuper; +import org.apache.fluss.memory.MemorySegment; +import org.apache.fluss.metadata.DataLakeFormat; +import org.apache.fluss.metadata.ResolvedPartitionSpec; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.decode.FixedSchemaDecoder; +import org.apache.fluss.row.encode.KeyEncoder; + +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.data.RowData; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.Map; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Runtime for blocking point lookups against a lake table. */ +final class LakeLookupRuntime implements Serializable { + + private static final long serialVersionUID = 1L; + + private final Configuration flussConfig; + private final TablePath tablePath; + private final org.apache.fluss.types.RowType flussFullRowType; + private final int[] primaryKeyIndexes; + private final Map tableOptions; + + @Nullable private transient LakeTableLookuper lakeTableLookuper; + @Nullable private transient FixedSchemaDecoder lakeValueDecoder; + @Nullable private transient KeyEncoder lakePrimaryKeyEncoder; + @Nullable private transient KeyEncoder lakeBucketKeyEncoder; + @Nullable private transient BucketingFunction bucketingFunction; + + @Nullable + private transient org.apache.fluss.client.table.getter.PartitionGetter partitionGetter; + + private transient short lakeSchemaId; + private transient int numBuckets; + + LakeLookupRuntime( + Configuration flussConfig, + TablePath tablePath, + org.apache.fluss.types.RowType flussFullRowType, + int[] primaryKeyIndexes, + Map tableOptions) { + this.flussConfig = checkNotNull(flussConfig, "flussConfig must not be null."); + this.tablePath = checkNotNull(tablePath, "tablePath must not be null."); + this.flussFullRowType = + checkNotNull(flussFullRowType, "flussFullRowType must not be null."); + this.primaryKeyIndexes = + checkNotNull(primaryKeyIndexes, "primaryKeyIndexes must not be null."); + this.tableOptions = checkNotNull(tableOptions, "tableOptions must not be null."); + } + + void open(TableInfo tableInfo) { + TableInfo resolvedTableInfo = checkNotNull(tableInfo, "tableInfo must not be null."); + DataLakeFormat dataLakeFormat = validateAndGetDataLakeFormat(resolvedTableInfo); + org.apache.fluss.types.RowType lookupRowType = flussFullRowType.project(primaryKeyIndexes); + lakePrimaryKeyEncoder = + KeyEncoder.ofPrimaryKeyEncoder( + lookupRowType, + resolvedTableInfo.getPhysicalPrimaryKeys(), + resolvedTableInfo.getTableConfig(), + resolvedTableInfo.isDefaultBucketKey()); + lakeBucketKeyEncoder = + KeyEncoder.ofBucketKeyEncoder( + lookupRowType, + resolvedTableInfo.getBucketKeys(), + resolvedTableInfo.getTableConfig(), + resolvedTableInfo.isDefaultBucketKey(), + lakePrimaryKeyEncoder); + bucketingFunction = BucketingFunction.of(dataLakeFormat); + partitionGetter = + new org.apache.fluss.client.table.getter.PartitionGetter( + lookupRowType, resolvedTableInfo.getPartitionKeys()); + numBuckets = resolvedTableInfo.getNumBuckets(); + lakeSchemaId = (short) resolvedTableInfo.getSchemaId(); + lakeValueDecoder = + new FixedSchemaDecoder( + resolvedTableInfo.getTableConfig().getKvFormat(), + resolvedTableInfo.getSchema()); + lakeTableLookuper = createLakeTableLookuper(dataLakeFormat, resolvedTableInfo); + } + + LakeLookupKey createLookupKey(RowData normalizedKeyRow) { + InternalRow lookupRow = new FlinkAsFlussRow(normalizedKeyRow); + KeyEncoder primaryKeyEncoder = + checkNotNull( + lakePrimaryKeyEncoder, "Lake primary-key encoder must be initialized."); + byte[] keyBytes = primaryKeyEncoder.encodeKey(lookupRow); + byte[] bucketKeyBytes = + lakeBucketKeyEncoder == primaryKeyEncoder + ? keyBytes + : checkNotNull( + lakeBucketKeyEncoder, + "Lake bucket-key encoder must be initialized.") + .encodeKey(lookupRow); + int bucketId = + checkNotNull(bucketingFunction, "Bucketing function must be initialized.") + .bucketing(bucketKeyBytes, numBuckets); + ResolvedPartitionSpec partitionSpec = + checkNotNull(partitionGetter, "Partition getter must be initialized.") + .getResolvedPartitionSpec(lookupRow); + LakeTableLookuper.LookupContext lookupContext = + new LakeTableLookuper.LookupContext( + partitionSpec, bucketId, lakeSchemaId, flussFullRowType); + return new LakeLookupKey(keyBytes, lookupContext); + } + + @Nullable + InternalRow lookup(LakeLookupKey lakeLookupKey) throws Exception { + byte[] value = + checkNotNull(lakeTableLookuper, "Lake table lookuper must be initialized.") + .lookup(lakeLookupKey.keyBytes, lakeLookupKey.lookupContext); + if (value == null) { + return null; + } + return checkNotNull(lakeValueDecoder, "Lake value decoder must be initialized.") + .decode(MemorySegment.wrap(value)); + } + + void close() throws Exception { + if (lakeTableLookuper != null) { + lakeTableLookuper.close(); + } + } + + private DataLakeFormat validateAndGetDataLakeFormat(TableInfo tableInfo) { + DataLakeFormat dataLakeFormat = + checkNotNull( + tableInfo.getTableConfig().getDataLakeFormat().orElse(null), + "Data lake format must be configured for lake fallback lookup."); + if (dataLakeFormat != DataLakeFormat.PAIMON) { + throw new TableException( + "Hybrid lake lookup currently only supports Paimon, but table " + + tablePath + + " uses " + + dataLakeFormat + + "."); + } + return dataLakeFormat; + } + + private LakeTableLookuper createLakeTableLookuper( + DataLakeFormat dataLakeFormat, TableInfo tableInfo) { + Configuration tableConfiguration = Configuration.fromMap(tableOptions); + Map lakeCatalogProperties = + DataLakeUtils.extractLakeCatalogProperties(tableConfiguration); + LakeStoragePlugin lakeStoragePlugin = + LakeStoragePluginSetUp.fromDataLakeFormat(dataLakeFormat.toString(), null); + LakeStorage lakeStorage = + checkNotNull(lakeStoragePlugin, "Lake storage plugin must not be null.") + .createLakeStorage(Configuration.fromMap(lakeCatalogProperties)); + return checkNotNull( + lakeStorage.createLakeTableLookuper( + tablePath, + new LakeStorage.LookuperContext( + flussConfig.get(ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR), + tableInfo.getTableConfig())), + "Lake table lookuper must not be null."); + } + + /** The encoded lake lookup key and its lake lookup context. */ + static final class LakeLookupKey { + private final byte[] keyBytes; + private final LakeTableLookuper.LookupContext lookupContext; + + private LakeLookupKey(byte[] keyBytes, LakeTableLookuper.LookupContext lookupContext) { + this.keyBytes = keyBytes; + this.lookupContext = lookupContext; + } + + ResolvedPartitionSpec getPartitionSpec() { + return lookupContext.partitionSpec(); + } + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java new file mode 100644 index 0000000000..e8e174774b --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.source.lookup; + +import org.apache.fluss.flink.utils.FlussRowToFlinkRowConverter; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.ProjectedRow; +import org.apache.fluss.types.RowType; + +import org.apache.flink.table.data.RowData; + +import javax.annotation.Nullable; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +/** Converts Fluss lookup results to projected and filtered Flink rows. */ +final class LookupResultConverter { + + private final int[] projection; + private final FlussRowToFlinkRowConverter rowConverter; + + LookupResultConverter(RowType outputRowType, int[] projection) { + this.projection = projection; + this.rowConverter = new FlussRowToFlinkRowConverter(outputRowType); + } + + Collection convert( + @Nullable List lookupRows, + @Nullable LookupNormalizer.RemainingFilter remainingFilter) { + if (lookupRows == null || lookupRows.isEmpty()) { + return Collections.emptyList(); + } + + List projectedRows = new ArrayList<>(lookupRows.size()); + // ProjectedRow must not be shared between concurrent async lookup requests. + ProjectedRow projectedRow = ProjectedRow.from(projection); + for (InternalRow row : lookupRows) { + if (row != null) { + RowData flinkRow = rowConverter.toFlinkRowData(projectedRow.replaceRow(row)); + if (remainingFilter == null || remainingFilter.isMatch(flinkRow)) { + projectedRows.add(flinkRow); + } + } + } + return projectedRows; + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java index 551d22ac23..9df91347a8 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java @@ -22,12 +22,14 @@ import org.apache.fluss.flink.source.FlinkTableSource; 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.flink.api.common.RuntimeExecutionMode; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.ExecutionOptions; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.catalog.CatalogTable; import org.apache.flink.table.catalog.Column; @@ -187,6 +189,78 @@ void testLookupSource() { .hasMessageContaining("Full lookup caching is not supported yet."); } + @Test + void testLakeFallbackLookupSource() { + ResolvedSchema schema = createPartitionedPkSchema(); + FlinkTableSource tableSource = + (FlinkTableSource) + createTableSource( + schema, getLakeFallbackOptions(), Collections.singletonList("pt")); + + LookupTableSource.LookupRuntimeProvider lookupProvider = + tableSource.getLookupRuntimeProvider( + new LookupRuntimeProviderContext(new int[][] {{0}, {1}, {2}})); + assertThat(lookupProvider).isInstanceOf(AsyncLookupFunctionProvider.class); + AsyncLookupFunction asyncLookupFunction = + ((AsyncLookupFunctionProvider) lookupProvider).createAsyncLookupFunction(); + assertThat(asyncLookupFunction).isInstanceOf(HybridLakeAsyncLookupFunction.class); + } + + @Test + void testLakeFallbackLookupSourceValidation() { + ResolvedSchema schema = createPartitionedPkSchema(); + + Map syncLookupProperties = getLakeFallbackOptions(); + syncLookupProperties.put(FlinkConnectorOptions.LOOKUP_ASYNC.key(), "false"); + assertThatThrownBy( + () -> + ((FlinkTableSource) + createTableSource( + schema, + syncLookupProperties, + Collections.singletonList("pt"))) + .getLookupRuntimeProvider( + new LookupRuntimeProviderContext( + new int[][] {{0}, {1}, {2}}))) + .isInstanceOf(TableException.class) + .hasMessageContaining( + "Option 'lookup.lake-fallback.enabled' requires 'lookup.async' to be true."); + + Map partialLookupProperties = getLakeFallbackOptions(); + partialLookupProperties.put("lookup.cache", "partial"); + partialLookupProperties.put(PARTIAL_CACHE_EXPIRE_AFTER_ACCESS.key(), "18000"); + partialLookupProperties.put(PARTIAL_CACHE_EXPIRE_AFTER_WRITE.key(), "36000"); + partialLookupProperties.put(PARTIAL_CACHE_MAX_ROWS.key(), "100000"); + assertThatThrownBy( + () -> + ((FlinkTableSource) + createTableSource( + schema, + partialLookupProperties, + Collections.singletonList("pt"))) + .getLookupRuntimeProvider( + new LookupRuntimeProviderContext( + new int[][] {{0}, {1}, {2}}))) + .isInstanceOf(TableException.class) + .hasMessageContaining( + "Option 'lookup.lake-fallback.enabled' cannot be used with lookup cache."); + + Map nonFullLookupProperties = getLakeFallbackOptions(); + assertThatThrownBy( + () -> + ((FlinkTableSource) + createTableSource( + schema, + nonFullLookupProperties, + Collections.singletonList("pt"))) + .getLookupRuntimeProvider( + new LookupRuntimeProviderContext( + new int[][] {{0}, {2}}))) + .isInstanceOf(TableException.class) + .hasMessageContaining( + "Option 'lookup.lake-fallback.enabled' only supports full primary-key lookup."); + } + @Test void testVirtualLogTableSourceDoesNotSupportBatchMode() { ResolvedSchema schema = createBasicSchema(); @@ -242,6 +316,18 @@ private ResolvedSchema createBasicSchema() { UniqueConstraint.primaryKey("PK_first_third", Arrays.asList("first", "third"))); } + private ResolvedSchema createPartitionedPkSchema() { + return new ResolvedSchema( + Arrays.asList( + Column.physical("id", DataTypes.INT().notNull()), + Column.physical("sub_id", DataTypes.INT().notNull()), + Column.physical("pt", DataTypes.STRING().notNull()), + Column.physical("value", DataTypes.STRING())), + Collections.emptyList(), + UniqueConstraint.primaryKey( + "PK_id_sub_id_pt", Arrays.asList("id", "sub_id", "pt"))); + } + private ResolvedSchema createBinlogSchema() { return new ResolvedSchema( Arrays.asList( @@ -277,11 +363,32 @@ private static Map getBasicOptionsWithBucketKey() { return basicOptions; } + private static Map getLakeFallbackOptions() { + Map options = getBasicOptions(); + options.put(BUCKET_KEY.key(), "id"); + options.put("table.datalake.enabled", "true"); + options.put("table.datalake.format", "paimon"); + options.put(FlinkConnectorOptions.LOOKUP_ASYNC.key(), "true"); + options.put(FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_ENABLED.key(), "true"); + return options; + } + private static DynamicTableSource createTableSource( ResolvedSchema schema, Map options) { return createTableSource(schema, options, Collections.emptyMap()); } + private static DynamicTableSource createTableSource( + ResolvedSchema schema, Map options, List partitionKeys) { + return createTableSource( + OBJECT_IDENTIFIER, + schema, + options, + Collections.emptyMap(), + new Configuration(), + partitionKeys); + } + private static DynamicTableSource createTableSource( ResolvedSchema schema, Map options, @@ -296,6 +403,24 @@ private static DynamicTableSource createTableSource( Map options, Map enrichmentOptions, Configuration configuration) { + return createTableSource( + objectIdentifier, + schema, + options, + enrichmentOptions, + configuration, + schema.getPrimaryKey() + .map(UniqueConstraint::getColumns) + .orElse(Collections.emptyList())); + } + + private static DynamicTableSource createTableSource( + ObjectIdentifier objectIdentifier, + ResolvedSchema schema, + Map options, + Map enrichmentOptions, + Configuration configuration, + List partitionKeys) { FlinkTableFactory tableFactory = createFlinkTableFactory(); FactoryUtil.DefaultDynamicTableContext context = new FactoryUtil.DefaultDynamicTableContext( @@ -304,9 +429,7 @@ private static DynamicTableSource createTableSource( CatalogTable.of( Schema.newBuilder().fromResolvedSchema(schema).build(), "mock source", - schema.getPrimaryKey() - .map(UniqueConstraint::getColumns) - .orElse(Collections.emptyList()), + partitionKeys, options), schema), enrichmentOptions, diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java new file mode 100644 index 0000000000..a8844f24db --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.lake.values; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.TableAlreadyExistException; +import org.apache.fluss.exception.TableNotExistException; +import org.apache.fluss.lake.lakestorage.LakeCatalog; +import org.apache.fluss.lake.lakestorage.LakeStorage; +import org.apache.fluss.lake.lakestorage.LakeStoragePlugin; +import org.apache.fluss.lake.serializer.SimpleVersionedSerializer; +import org.apache.fluss.lake.source.LakeSource; +import org.apache.fluss.lake.source.LakeSplit; +import org.apache.fluss.lake.source.Planner; +import org.apache.fluss.lake.source.RecordReader; +import org.apache.fluss.lake.writer.LakeTieringFactory; +import org.apache.fluss.metadata.DataLakeFormat; +import org.apache.fluss.metadata.TableChange; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.predicate.Predicate; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; + +/** Test-only Paimon lake storage plugin used to construct Flink table sources. */ +public class TestingPaimonLakeStoragePlugin implements LakeStoragePlugin { + + @Override + public String identifier() { + return DataLakeFormat.PAIMON.toString(); + } + + @Override + public LakeStorage createLakeStorage(Configuration configuration) { + return new TestingPaimonLakeStorage(); + } + + private static class TestingPaimonLakeStorage implements LakeStorage { + @Override + public LakeTieringFactory createLakeTieringFactory() { + throw new UnsupportedOperationException("Not implemented."); + } + + @Override + public LakeCatalog createLakeCatalog() { + return new TestingPaimonLakeCatalog(); + } + + @Override + public LakeSource createLakeSource(TablePath tablePath) { + return new TestingPaimonLakeSource(); + } + } + + private static class TestingPaimonLakeCatalog implements LakeCatalog { + @Override + public void createTable( + TablePath tablePath, TableDescriptor tableDescriptor, Context context) + throws TableAlreadyExistException {} + + @Override + public void alterTable(TablePath tablePath, List tableChanges, Context context) + throws TableNotExistException {} + } + + private static class TestingPaimonLakeSource implements LakeSource { + @Override + public void withProject(int[][] project) {} + + @Override + public void withLimit(int limit) {} + + @Override + public FilterPushDownResult withFilters(List predicates) { + return FilterPushDownResult.of(predicates, Collections.emptyList()); + } + + @Override + public Planner createPlanner(PlannerContext context) throws IOException { + return Collections::emptyList; + } + + @Override + public RecordReader createRecordReader(ReaderContext context) { + throw new UnsupportedOperationException("Not implemented."); + } + + @Override + public SimpleVersionedSerializer getSplitSerializer() { + throw new UnsupportedOperationException("Not implemented."); + } + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/resources/META-INF/services/org.apache.fluss.lake.lakestorage.LakeStoragePlugin b/fluss-flink/fluss-flink-common/src/test/resources/META-INF/services/org.apache.fluss.lake.lakestorage.LakeStoragePlugin index 3fafafad01..1497369939 100644 --- a/fluss-flink/fluss-flink-common/src/test/resources/META-INF/services/org.apache.fluss.lake.lakestorage.LakeStoragePlugin +++ b/fluss-flink/fluss-flink-common/src/test/resources/META-INF/services/org.apache.fluss.lake.lakestorage.LakeStoragePlugin @@ -16,4 +16,5 @@ # limitations under the License. # -org.apache.fluss.lake.values.TestingValuesLakeStoragePlugin \ No newline at end of file +org.apache.fluss.lake.values.TestingValuesLakeStoragePlugin +org.apache.fluss.lake.values.TestingPaimonLakeStoragePlugin From 7f083b917bc2738aaf50c490e20d36d3ede9f453 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Fri, 7 Aug 2026 15:13:03 +0800 Subject: [PATCH 2/6] cache involved --- .../fluss/flink/source/FlinkTableSource.java | 5 +- .../lookup/FlinkAsyncLookupFunction.java | 20 +- .../source/lookup/FlinkLookupFunction.java | 20 +- .../lookup/HybridLakeAsyncLookupFunction.java | 173 ++++++++++-------- .../source/lookup/LookupResultConverter.java | 19 +- .../flink/catalog/FlinkTableFactoryTest.java | 17 ++ 6 files changed, 145 insertions(+), 109 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java index 7a8653d7fd..8e62c37917 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java @@ -549,7 +549,6 @@ public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { tablePath, tableOutputType, primaryKeyIndexes, - partitionKeyIndexes, lookupNormalizer, projectedFields, tableOptions, @@ -619,6 +618,10 @@ private void validateLakeFallbackLookup(LookupNormalizer lookupNormalizer) { 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 " diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java index 437ac72fe9..96440cf04e 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java @@ -47,9 +47,7 @@ public class FlinkAsyncLookupFunction extends AsyncLookupFunction { private static final long serialVersionUID = 1L; private final TablePath tablePath; - private final RowType flinkRowType; private final LookupNormalizer lookupNormalizer; - private final int[] projection; private final FlussLookupRuntime flussLookupRuntime; private transient LookupResultConverter lookupResultConverter; @@ -61,25 +59,25 @@ public FlinkAsyncLookupFunction( @Nullable int[] projection, boolean insertIfNotExists) { this.tablePath = tablePath; - this.flinkRowType = flinkRowType; this.lookupNormalizer = lookupNormalizer; - this.projection = - projection == null - ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() - : projection; this.flussLookupRuntime = new FlussLookupRuntime( flussConfig, tablePath, flinkRowType, lookupNormalizer, insertIfNotExists); + + int[] resolvedProjection = + projection == null + ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() + : projection; + RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, resolvedProjection); + lookupResultConverter = + new LookupResultConverter( + FlinkConversions.toFlussRowType(outputRowType), resolvedProjection); } @Override public void open(FunctionContext context) { LOG.info("start open ..."); flussLookupRuntime.open(); - RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, projection); - lookupResultConverter = - new LookupResultConverter( - FlinkConversions.toFlussRowType(outputRowType), projection); LOG.info("end open."); } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java index e9b051a84e..6f35fa3f1a 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java @@ -44,9 +44,7 @@ public class FlinkLookupFunction extends LookupFunction { private static final Logger LOG = LoggerFactory.getLogger(FlinkLookupFunction.class); private static final long serialVersionUID = 1L; - private final RowType flinkRowType; private final LookupNormalizer lookupNormalizer; - private final int[] projection; private final FlussLookupRuntime flussLookupRuntime; private transient LookupResultConverter lookupResultConverter; @@ -57,25 +55,25 @@ public FlinkLookupFunction( LookupNormalizer lookupNormalizer, @Nullable int[] projection, boolean insertIfNotExists) { - this.flinkRowType = flinkRowType; this.lookupNormalizer = lookupNormalizer; - this.projection = - projection == null - ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() - : projection; this.flussLookupRuntime = new FlussLookupRuntime( flussConfig, tablePath, flinkRowType, lookupNormalizer, insertIfNotExists); + + int[] resolvedProjection = + projection == null + ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() + : projection; + RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, resolvedProjection); + this.lookupResultConverter = + new LookupResultConverter( + FlinkConversions.toFlussRowType(outputRowType), resolvedProjection); } @Override public void open(FunctionContext context) { LOG.info("start open ..."); flussLookupRuntime.open(); - RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, projection); - lookupResultConverter = - new LookupResultConverter( - FlinkConversions.toFlussRowType(outputRowType), projection); LOG.info("end open."); } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java index 06815903cf..27b3b13438 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java @@ -18,18 +18,18 @@ package org.apache.fluss.flink.source.lookup; import org.apache.fluss.client.admin.Admin; -import org.apache.fluss.client.lookup.LookupType; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.PartitionNotExistException; import org.apache.fluss.flink.utils.FlinkConversions; import org.apache.fluss.flink.utils.FlinkUtils; +import org.apache.fluss.metadata.PartitionInfo; +import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.row.InternalRow; import org.apache.fluss.utils.ExceptionUtils; import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; -import org.apache.flink.table.api.TableException; import org.apache.flink.table.data.RowData; import org.apache.flink.table.functions.AsyncLookupFunction; import org.apache.flink.table.functions.FunctionContext; @@ -42,10 +42,13 @@ import java.time.Duration; import java.util.Collection; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; @@ -56,8 +59,6 @@ import java.util.concurrent.TimeoutException; import java.util.stream.IntStream; -import static org.apache.fluss.utils.Preconditions.checkNotNull; - /** * An async lookup function that first looks up Fluss and falls back to a lake point lookup when the * requested Fluss partition is absent. @@ -73,7 +74,10 @@ public class HybridLakeAsyncLookupFunction extends AsyncLookupFunction { private final TablePath tablePath; private final LookupNormalizer lookupNormalizer; private final FlussLookupRuntime flussLookupRuntime; - private transient LookupResultConverter lookupResultConverter; + private final LookupResultConverter lookupResultConverter; + + // Auto-created partitions cached as absent are expired and will never become live again. + private final Map partitionExistenceCache = new ConcurrentHashMap<>(); private final LakeLookupRuntime lakeLookupRuntime; private final Duration lakeFallbackTimeout; @@ -81,13 +85,13 @@ public class HybridLakeAsyncLookupFunction extends AsyncLookupFunction { private final int lakeFallbackMaxConcurrency; private transient ThreadPoolExecutor lakeLookupExecutor; private transient ScheduledExecutorService timeoutExecutor; + private transient Admin admin; public HybridLakeAsyncLookupFunction( Configuration flussConfig, TablePath tablePath, RowType flinkRowType, int[] primaryKeyIndexes, - int[] partitionKeyIndexes, LookupNormalizer lookupNormalizer, @Nullable int[] projection, Map tableOptions, @@ -101,14 +105,6 @@ public HybridLakeAsyncLookupFunction( this.lakeFallbackExecutorThreads = lakeFallbackExecutorThreads; this.lakeFallbackMaxConcurrency = lakeFallbackMaxConcurrency; - validateLookupShape( - primaryKeyIndexes, - partitionKeyIndexes, - lookupNormalizer, - lakeFallbackTimeout, - lakeFallbackExecutorThreads, - lakeFallbackMaxConcurrency); - int[] resolvedProjection = projection == null ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() @@ -135,6 +131,8 @@ public HybridLakeAsyncLookupFunction( public void open(@Nullable FunctionContext context) { LOG.info("Starting hybrid lake async lookup function for table {}.", tablePath); flussLookupRuntime.open(); + admin = flussLookupRuntime.getAdmin(); + initializePartitionExistenceCache(); TableInfo tableInfo = flussLookupRuntime.getTableInfo(); lakeLookupRuntime.open(tableInfo); @@ -146,6 +144,47 @@ public void open(@Nullable FunctionContext context) { LOG.info("Finished opening hybrid lake async lookup function for table {}.", tablePath); } + private void initializePartitionExistenceCache() { + partitionExistenceCache.clear(); + try { + List partitionInfos = admin.listPartitionInfos(tablePath).get(); + for (PartitionInfo partitionInfo : partitionInfos) { + partitionExistenceCache.put(partitionInfo.getPartitionSpec(), true); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException( + "Interrupted while initializing partitions for table " + tablePath + ".", e); + } catch (ExecutionException e) { + throw new RuntimeException( + "Failed to initialize partitions for table " + tablePath + ".", e.getCause()); + } + } + + private synchronized boolean getOrRefreshPartitionExistence(PartitionSpec partitionSpec) { + Boolean partitionExists = partitionExistenceCache.get(partitionSpec); + if (partitionExists != null) { + return partitionExists; + } + try { + partitionExists = !admin.listPartitionInfos(tablePath, partitionSpec).get().isEmpty(); + partitionExistenceCache.put(partitionSpec, partitionExists); + return partitionExists; + } catch (Exception e) { + throw new RuntimeException( + "Failed to refresh partition " + + partitionSpec + + " for table " + + tablePath + + ".", + e.getCause()); + } + } + + private void markPartitionInvalid(PartitionSpec partitionSpec) { + partitionExistenceCache.put(partitionSpec, false); + } + @Override public CompletableFuture> asyncLookup(RowData keyRow) { RowData normalizedKeyRow = lookupNormalizer.normalizeLookupKey(keyRow); @@ -155,6 +194,37 @@ public CompletableFuture> asyncLookup(RowData keyRow) { lakeLookupRuntime.createLookupKey(normalizedKeyRow); CompletableFuture> future = new CompletableFuture<>(); + lookupByPartition(normalizedKeyRow, lakeLookupKey, remainingFilter, future); + return future; + } + + private void lookupByPartition( + RowData normalizedKeyRow, + LakeLookupRuntime.LakeLookupKey lakeLookupKey, + @Nullable LookupNormalizer.RemainingFilter remainingFilter, + CompletableFuture> future) { + try { + PartitionSpec partitionSpec = lakeLookupKey.getPartitionSpec().toPartitionSpec(); + Boolean partitionExists = partitionExistenceCache.get(partitionSpec); + if (partitionExists == null) { + partitionExists = getOrRefreshPartitionExistence(partitionSpec); + } + if (partitionExists) { + lookupFlussAsync(normalizedKeyRow, lakeLookupKey, remainingFilter, future); + } else { + lookupLakeAsync(lakeLookupKey, remainingFilter, future); + } + } catch (Throwable t) { + future.completeExceptionally(t); + } + } + + private void lookupFlussAsync( + RowData normalizedKeyRow, + LakeLookupRuntime.LakeLookupKey lakeLookupKey, + @Nullable LookupNormalizer.RemainingFilter remainingFilter, + CompletableFuture> future) { + PartitionSpec partitionSpec = lakeLookupKey.getPartitionSpec().toPartitionSpec(); try { flussLookupRuntime .lookup(normalizedKeyRow) @@ -165,8 +235,8 @@ public CompletableFuture> asyncLookup(RowData keyRow) { if (ExceptionUtils.findThrowable( throwable, PartitionNotExistException.class) .isPresent()) { - checkPartitionAndLookupLake( - lakeLookupKey, remainingFilter, future); + markPartitionInvalid(partitionSpec); + lookupLakeAsync(lakeLookupKey, remainingFilter, future); return; } LOG.error( @@ -181,18 +251,12 @@ public CompletableFuture> asyncLookup(RowData keyRow) { return; } - boolean hit = result != null && !result.getRowList().isEmpty(); - if (hit) { + if (result != null && !result.getRowList().isEmpty()) { future.complete( - checkNotNull( - lookupResultConverter, - "Lookup result converter is not initialized.") - .convert( - result.getRowList(), - remainingFilter)); + lookupResultConverter.convert( + result.getRowList(), remainingFilter)); } else { - checkPartitionAndLookupLake( - lakeLookupKey, remainingFilter, future); + future.complete(Collections.emptyList()); } } catch (Throwable t) { future.completeExceptionally(t); @@ -200,37 +264,12 @@ public CompletableFuture> asyncLookup(RowData keyRow) { }); } catch (Throwable t) { if (ExceptionUtils.findThrowable(t, PartitionNotExistException.class).isPresent()) { - checkPartitionAndLookupLake(lakeLookupKey, remainingFilter, future); + markPartitionInvalid(partitionSpec); + lookupLakeAsync(lakeLookupKey, remainingFilter, future); } else { future.completeExceptionally(t); } } - return future; - } - - private void checkPartitionAndLookupLake( - LakeLookupRuntime.LakeLookupKey lakeLookupKey, - @Nullable LookupNormalizer.RemainingFilter remainingFilter, - CompletableFuture> future) { - Admin admin = flussLookupRuntime.getAdmin(); - admin.listPartitionInfos(tablePath, lakeLookupKey.getPartitionSpec().toPartitionSpec()) - .whenComplete( - (partitionInfos, throwable) -> { - try { - if (throwable != null) { - future.completeExceptionally(throwable); - return; - } - boolean partitionExists = !partitionInfos.isEmpty(); - if (partitionExists) { - future.complete(Collections.emptyList()); - } else { - lookupLakeAsync(lakeLookupKey, remainingFilter, future); - } - } catch (Throwable t) { - future.completeExceptionally(t); - } - }); } private ThreadPoolExecutor createLakeLookupExecutor() { @@ -315,34 +354,6 @@ private Collection lookupLake( return lookupResultConverter.convert(Collections.singletonList(row), remainingFilter); } - private static void validateLookupShape( - int[] primaryKeyIndexes, - int[] partitionKeyIndexes, - LookupNormalizer lookupNormalizer, - Duration lakeFallbackTimeout, - int lakeFallbackExecutorThreads, - int lakeFallbackMaxConcurrency) { - if (primaryKeyIndexes.length == 0) { - throw new TableException("Lake fallback lookup requires a primary-key table."); - } - if (partitionKeyIndexes.length == 0) { - throw new TableException("Lake fallback lookup requires a partitioned table."); - } - if (lookupNormalizer.getLookupType() != LookupType.LOOKUP) { - throw new TableException("Lake fallback lookup only supports full primary-key lookup."); - } - if (lakeFallbackTimeout.isZero() || lakeFallbackTimeout.isNegative()) { - throw new TableException("Lake fallback lookup timeout must be positive."); - } - if (lakeFallbackExecutorThreads <= 0 || lakeFallbackMaxConcurrency <= 0) { - throw new TableException("Lake fallback lookup executor settings must be positive."); - } - if (lakeFallbackExecutorThreads > lakeFallbackMaxConcurrency) { - throw new TableException( - "Lake fallback lookup executor threads must not exceed max concurrency."); - } - } - @Override public void close() throws Exception { LOG.info("Closing hybrid lake async lookup function for table {}.", tablePath); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java index e8e174774b..3da1e07b05 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupResultConverter.java @@ -26,16 +26,19 @@ import javax.annotation.Nullable; +import java.io.Serializable; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.List; /** Converts Fluss lookup results to projected and filtered Flink rows. */ -final class LookupResultConverter { +final class LookupResultConverter implements Serializable { + + private static final long serialVersionUID = 1L; private final int[] projection; - private final FlussRowToFlinkRowConverter rowConverter; + private transient volatile FlussRowToFlinkRowConverter rowConverter; LookupResultConverter(RowType outputRowType, int[] projection) { this.projection = projection; @@ -50,11 +53,9 @@ Collection convert( } List projectedRows = new ArrayList<>(lookupRows.size()); - // ProjectedRow must not be shared between concurrent async lookup requests. - ProjectedRow projectedRow = ProjectedRow.from(projection); for (InternalRow row : lookupRows) { if (row != null) { - RowData flinkRow = rowConverter.toFlinkRowData(projectedRow.replaceRow(row)); + RowData flinkRow = rowConverter.toFlinkRowData(maybeProject(row)); if (remainingFilter == null || remainingFilter.isMatch(flinkRow)) { projectedRows.add(flinkRow); } @@ -62,4 +63,12 @@ Collection convert( } return projectedRows; } + + private InternalRow maybeProject(InternalRow row) { + if (projection == null) { + return row; + } + // should not reuse objects for async operations + return ProjectedRow.from(projection).replaceRow(row); + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java index 9df91347a8..d43192aad8 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java @@ -259,6 +259,22 @@ void testLakeFallbackLookupSourceValidation() { .isInstanceOf(TableException.class) .hasMessageContaining( "Option 'lookup.lake-fallback.enabled' only supports full primary-key lookup."); + + Map nonAutoPartitionProperties = getLakeFallbackOptions(); + nonAutoPartitionProperties.put("table.auto-partition.enabled", "false"); + assertThatThrownBy( + () -> + ((FlinkTableSource) + createTableSource( + schema, + nonAutoPartitionProperties, + Collections.singletonList("pt"))) + .getLookupRuntimeProvider( + new LookupRuntimeProviderContext( + new int[][] {{0}, {1}, {2}}))) + .isInstanceOf(TableException.class) + .hasMessageContaining( + "Option 'lookup.lake-fallback.enabled' requires an auto-partitioned table."); } @Test @@ -368,6 +384,7 @@ private static Map getLakeFallbackOptions() { options.put(BUCKET_KEY.key(), "id"); options.put("table.datalake.enabled", "true"); options.put("table.datalake.format", "paimon"); + options.put("table.auto-partition.enabled", "true"); options.put(FlinkConnectorOptions.LOOKUP_ASYNC.key(), "true"); options.put(FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_ENABLED.key(), "true"); return options; From 8b9876999e579ca0d67fe9f3eca0fc49b13e4939 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Fri, 7 Aug 2026 15:21:09 +0800 Subject: [PATCH 3/6] fix --- .../fluss/flink/source/lookup/FlinkAsyncLookupFunction.java | 4 ++-- .../apache/fluss/flink/source/lookup/FlinkLookupFunction.java | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java index 96440cf04e..7b08d53525 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java @@ -49,7 +49,7 @@ public class FlinkAsyncLookupFunction extends AsyncLookupFunction { private final TablePath tablePath; private final LookupNormalizer lookupNormalizer; private final FlussLookupRuntime flussLookupRuntime; - private transient LookupResultConverter lookupResultConverter; + private final LookupResultConverter lookupResultConverter; public FlinkAsyncLookupFunction( Configuration flussConfig, @@ -69,7 +69,7 @@ public FlinkAsyncLookupFunction( ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() : projection; RowType outputRowType = FlinkUtils.projectRowType(flinkRowType, resolvedProjection); - lookupResultConverter = + this.lookupResultConverter = new LookupResultConverter( FlinkConversions.toFlussRowType(outputRowType), resolvedProjection); } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java index 6f35fa3f1a..9606c8e632 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlinkLookupFunction.java @@ -46,7 +46,7 @@ public class FlinkLookupFunction extends LookupFunction { private final LookupNormalizer lookupNormalizer; private final FlussLookupRuntime flussLookupRuntime; - private transient LookupResultConverter lookupResultConverter; + private final LookupResultConverter lookupResultConverter; public FlinkLookupFunction( Configuration flussConfig, From 8f08083b7cd209ba12e44633d4958c24aac3c746 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Fri, 7 Aug 2026 15:58:06 +0800 Subject: [PATCH 4/6] add better abstraction --- .../source/lookup/FlussLookupRuntime.java | 26 +-- .../lookup/HybridLakeAsyncLookupFunction.java | 185 ++++------------ .../source/lookup/LakeLookupRuntime.java | 200 +++++++++++------- .../flink/source/lookup/LookupRuntime.java | 36 ++++ 4 files changed, 218 insertions(+), 229 deletions(-) create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupRuntime.java diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java index 1e363aa0ca..a84c52059a 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/FlussLookupRuntime.java @@ -39,10 +39,8 @@ import java.io.Serializable; import java.util.concurrent.CompletableFuture; -import static org.apache.fluss.utils.Preconditions.checkNotNull; - /** The shared Fluss client-side lookup runtime. */ -final class FlussLookupRuntime implements Serializable { +final class FlussLookupRuntime implements LookupRuntime, Serializable { private static final Logger LOG = LoggerFactory.getLogger(FlussLookupRuntime.class); @@ -54,9 +52,9 @@ final class FlussLookupRuntime implements Serializable { private final LookupNormalizer lookupNormalizer; private final boolean insertIfNotExists; - private transient Connection connection; - private transient Table table; - private transient Lookuper lookuper; + private Connection connection; + private Table table; + private Lookuper lookuper; FlussLookupRuntime( Configuration flussConfig, @@ -71,7 +69,8 @@ final class FlussLookupRuntime implements Serializable { this.insertIfNotExists = insertIfNotExists; } - void open() { + @Override + public void open() { LOG.info("Starting Fluss lookup runtime for table {}.", tablePath); connection = ConnectionFactory.createConnection(flussConfig); table = connection.getTable(tablePath); @@ -88,20 +87,21 @@ void open() { LOG.info("Finished starting Fluss lookup runtime."); } - CompletableFuture lookup(RowData normalizedKeyRow) { - return checkNotNull(lookuper, "Fluss lookuper must be initialized.") - .lookup(new FlinkAsFlussRow(normalizedKeyRow)); + @Override + public CompletableFuture lookup(RowData normalizedKeyRow) { + return lookuper.lookup(new FlinkAsFlussRow(normalizedKeyRow)); } TableInfo getTableInfo() { - return checkNotNull(table, "Fluss table must be initialized.").getTableInfo(); + return table.getTableInfo(); } Admin getAdmin() { - return checkNotNull(connection, "Fluss connection must be initialized.").getAdmin(); + return connection.getAdmin(); } - void close() throws Exception { + @Override + public void close() throws Exception { LOG.info("Closing Fluss lookup runtime for table {}.", tablePath); if (table != null) { table.close(); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java index 27b3b13438..427fb567af 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java @@ -18,17 +18,18 @@ package org.apache.fluss.flink.source.lookup; import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.table.getter.PartitionGetter; import org.apache.fluss.config.Configuration; import org.apache.fluss.exception.PartitionNotExistException; +import org.apache.fluss.flink.row.FlinkAsFlussRow; import org.apache.fluss.flink.utils.FlinkConversions; import org.apache.fluss.flink.utils.FlinkUtils; import org.apache.fluss.metadata.PartitionInfo; import org.apache.fluss.metadata.PartitionSpec; import org.apache.fluss.metadata.TableInfo; import org.apache.fluss.metadata.TablePath; -import org.apache.fluss.row.InternalRow; import org.apache.fluss.utils.ExceptionUtils; -import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; +import org.apache.fluss.utils.concurrent.FutureUtils; import org.apache.flink.table.data.RowData; import org.apache.flink.table.functions.AsyncLookupFunction; @@ -44,19 +45,9 @@ import java.util.Collections; import java.util.List; import java.util.Map; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; -import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.stream.IntStream; /** @@ -74,18 +65,15 @@ public class HybridLakeAsyncLookupFunction extends AsyncLookupFunction { private final TablePath tablePath; private final LookupNormalizer lookupNormalizer; private final FlussLookupRuntime flussLookupRuntime; + private final org.apache.fluss.types.RowType flussLookupRowType; private final LookupResultConverter lookupResultConverter; // Auto-created partitions cached as absent are expired and will never become live again. private final Map partitionExistenceCache = new ConcurrentHashMap<>(); private final LakeLookupRuntime lakeLookupRuntime; - private final Duration lakeFallbackTimeout; - private final int lakeFallbackExecutorThreads; - private final int lakeFallbackMaxConcurrency; - private transient ThreadPoolExecutor lakeLookupExecutor; - private transient ScheduledExecutorService timeoutExecutor; - private transient Admin admin; + private Admin admin; + private PartitionGetter partitionGetter; public HybridLakeAsyncLookupFunction( Configuration flussConfig, @@ -101,10 +89,6 @@ public HybridLakeAsyncLookupFunction( this.tablePath = tablePath; this.lookupNormalizer = lookupNormalizer; - this.lakeFallbackTimeout = lakeFallbackTimeout; - this.lakeFallbackExecutorThreads = lakeFallbackExecutorThreads; - this.lakeFallbackMaxConcurrency = lakeFallbackMaxConcurrency; - int[] resolvedProjection = projection == null ? IntStream.range(0, flinkRowType.getFieldCount()).toArray() @@ -114,33 +98,36 @@ public HybridLakeAsyncLookupFunction( FlinkConversions.toFlussRowType( FlinkUtils.projectRowType(flinkRowType, resolvedProjection)), resolvedProjection); + org.apache.fluss.types.RowType flussFullRowType = + FlinkConversions.toFlussRowType(flinkRowType); + this.flussLookupRowType = flussFullRowType.project(primaryKeyIndexes); this.flussLookupRuntime = new FlussLookupRuntime( flussConfig, tablePath, flinkRowType, lookupNormalizer, false); this.lakeLookupRuntime = new LakeLookupRuntime( + flussLookupRuntime, flussConfig, tablePath, - FlinkConversions.toFlussRowType(flinkRowType), + flussFullRowType, primaryKeyIndexes, - tableOptions); + tableOptions, + lakeFallbackTimeout, + lakeFallbackExecutorThreads, + lakeFallbackMaxConcurrency); } @Override public void open(@Nullable FunctionContext context) { LOG.info("Starting hybrid lake async lookup function for table {}.", tablePath); flussLookupRuntime.open(); + TableInfo tableInfo = flussLookupRuntime.getTableInfo(); + partitionGetter = new PartitionGetter(flussLookupRowType, tableInfo.getPartitionKeys()); admin = flussLookupRuntime.getAdmin(); initializePartitionExistenceCache(); - TableInfo tableInfo = flussLookupRuntime.getTableInfo(); - lakeLookupRuntime.open(tableInfo); - - lakeLookupExecutor = createLakeLookupExecutor(); - timeoutExecutor = - new ScheduledThreadPoolExecutor( - 1, new ExecutorThreadFactory("fluss-lake-fallback-timeout")); + lakeLookupRuntime.open(); LOG.info("Finished opening hybrid lake async lookup function for table {}.", tablePath); } @@ -190,41 +177,30 @@ public CompletableFuture> asyncLookup(RowData keyRow) { RowData normalizedKeyRow = lookupNormalizer.normalizeLookupKey(keyRow); LookupNormalizer.RemainingFilter remainingFilter = lookupNormalizer.createRemainingFilter(keyRow); - LakeLookupRuntime.LakeLookupKey lakeLookupKey = - lakeLookupRuntime.createLookupKey(normalizedKeyRow); - - CompletableFuture> future = new CompletableFuture<>(); - lookupByPartition(normalizedKeyRow, lakeLookupKey, remainingFilter, future); - return future; - } - - private void lookupByPartition( - RowData normalizedKeyRow, - LakeLookupRuntime.LakeLookupKey lakeLookupKey, - @Nullable LookupNormalizer.RemainingFilter remainingFilter, - CompletableFuture> future) { try { - PartitionSpec partitionSpec = lakeLookupKey.getPartitionSpec().toPartitionSpec(); + PartitionSpec partitionSpec = + partitionGetter + .getResolvedPartitionSpec(new FlinkAsFlussRow(normalizedKeyRow)) + .toPartitionSpec(); Boolean partitionExists = partitionExistenceCache.get(partitionSpec); if (partitionExists == null) { partitionExists = getOrRefreshPartitionExistence(partitionSpec); } if (partitionExists) { - lookupFlussAsync(normalizedKeyRow, lakeLookupKey, remainingFilter, future); + return lookupFlussAsync(normalizedKeyRow, partitionSpec, remainingFilter); } else { - lookupLakeAsync(lakeLookupKey, remainingFilter, future); + return lookupLakeAsync(normalizedKeyRow, remainingFilter); } } catch (Throwable t) { - future.completeExceptionally(t); + return FutureUtils.completedExceptionally(t); } } - private void lookupFlussAsync( + private CompletableFuture> lookupFlussAsync( RowData normalizedKeyRow, - LakeLookupRuntime.LakeLookupKey lakeLookupKey, - @Nullable LookupNormalizer.RemainingFilter remainingFilter, - CompletableFuture> future) { - PartitionSpec partitionSpec = lakeLookupKey.getPartitionSpec().toPartitionSpec(); + PartitionSpec partitionSpec, + @Nullable LookupNormalizer.RemainingFilter remainingFilter) { + CompletableFuture> future = new CompletableFuture<>(); try { flussLookupRuntime .lookup(normalizedKeyRow) @@ -236,7 +212,11 @@ private void lookupFlussAsync( throwable, PartitionNotExistException.class) .isPresent()) { markPartitionInvalid(partitionSpec); - lookupLakeAsync(lakeLookupKey, remainingFilter, future); + lookupLakeAsync(normalizedKeyRow, remainingFilter) + .whenComplete( + (rows, error) -> + FutureUtils.doForward( + rows, error, future)); return; } LOG.error( @@ -265,104 +245,27 @@ private void lookupFlussAsync( } catch (Throwable t) { if (ExceptionUtils.findThrowable(t, PartitionNotExistException.class).isPresent()) { markPartitionInvalid(partitionSpec); - lookupLakeAsync(lakeLookupKey, remainingFilter, future); + return lookupLakeAsync(normalizedKeyRow, remainingFilter); } else { future.completeExceptionally(t); } } + return future; } - private ThreadPoolExecutor createLakeLookupExecutor() { - int queueCapacity = lakeFallbackMaxConcurrency - lakeFallbackExecutorThreads; - BlockingQueue queue = - queueCapacity == 0 - ? new SynchronousQueue<>() - : new ArrayBlockingQueue<>(queueCapacity); - return new ThreadPoolExecutor( - lakeFallbackExecutorThreads, - lakeFallbackExecutorThreads, - 0L, - TimeUnit.MILLISECONDS, - queue, - new ExecutorThreadFactory("fluss-lake-fallback-lookup"), - new ThreadPoolExecutor.AbortPolicy()); - } - - private void lookupLakeAsync( - LakeLookupRuntime.LakeLookupKey lakeLookupKey, - @Nullable LookupNormalizer.RemainingFilter remainingFilter, - CompletableFuture> future) { - ScheduledFuture timeoutTask; - try { - timeoutTask = - timeoutExecutor.schedule( - () -> - completeLakeFallbackExceptionally( - future, - new TimeoutException( - "Lake fallback lookup timed out after " - + lakeFallbackTimeout)), - lakeFallbackTimeout.toMillis(), - TimeUnit.MILLISECONDS); - future.whenComplete((ignored, ignoredError) -> timeoutTask.cancel(false)); - } catch (RejectedExecutionException e) { - completeLakeFallbackExceptionally( - future, new RuntimeException("Lake fallback timeout executor is closed.", e)); - return; - } - - try { - lakeLookupExecutor.execute( - () -> { - try { - Collection rows = lookupLake(lakeLookupKey, remainingFilter); - completeLakeFallbackSuccessfully(future, rows); - } catch (Throwable t) { - completeLakeFallbackExceptionally( - future, - new RuntimeException( - "Execution of lake fallback lookup failed: " - + t.getMessage(), - t)); - } - }); - } catch (RejectedExecutionException e) { - completeLakeFallbackExceptionally( - future, - new RuntimeException("Lake fallback lookup executor is overloaded.", e)); - } - } - - private void completeLakeFallbackSuccessfully( - CompletableFuture> future, Collection rows) { - future.complete(rows); - } - - private void completeLakeFallbackExceptionally( - CompletableFuture> future, Throwable throwable) { - future.completeExceptionally(throwable); - } - - private Collection lookupLake( - LakeLookupRuntime.LakeLookupKey lakeLookupKey, - @Nullable LookupNormalizer.RemainingFilter remainingFilter) - throws Exception { - InternalRow row = lakeLookupRuntime.lookup(lakeLookupKey); - if (row == null) { - return Collections.emptyList(); - } - return lookupResultConverter.convert(Collections.singletonList(row), remainingFilter); + private CompletableFuture> lookupLakeAsync( + RowData normalizedKeyRow, @Nullable LookupNormalizer.RemainingFilter remainingFilter) { + return lakeLookupRuntime + .lookup(normalizedKeyRow) + .thenApply( + result -> + lookupResultConverter.convert( + result.getRowList(), remainingFilter)); } @Override public void close() throws Exception { LOG.info("Closing hybrid lake async lookup function for table {}.", tablePath); - if (lakeLookupExecutor != null) { - lakeLookupExecutor.shutdownNow(); - } - if (timeoutExecutor != null) { - timeoutExecutor.shutdownNow(); - } lakeLookupRuntime.close(); flussLookupRuntime.close(); } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java index 283c2f158a..056bf8185a 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LakeLookupRuntime.java @@ -18,6 +18,8 @@ package org.apache.fluss.flink.source.lookup; import org.apache.fluss.bucketing.BucketingFunction; +import org.apache.fluss.client.lookup.LookupResult; +import org.apache.fluss.client.table.getter.PartitionGetter; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.row.FlinkAsFlussRow; @@ -34,57 +36,75 @@ import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.decode.FixedSchemaDecoder; import org.apache.fluss.row.encode.KeyEncoder; +import org.apache.fluss.utils.concurrent.ExecutorThreadFactory; +import org.apache.fluss.utils.concurrent.FutureUtils; import org.apache.flink.table.api.TableException; import org.apache.flink.table.data.RowData; -import javax.annotation.Nullable; - import java.io.Serializable; +import java.time.Duration; import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; import static org.apache.fluss.utils.Preconditions.checkNotNull; -/** Runtime for blocking point lookups against a lake table. */ -final class LakeLookupRuntime implements Serializable { +/** Runtime for asynchronous point lookups against a lake table. */ +final class LakeLookupRuntime implements LookupRuntime, Serializable { private static final long serialVersionUID = 1L; + private final FlussLookupRuntime flussLookupRuntime; private final Configuration flussConfig; private final TablePath tablePath; private final org.apache.fluss.types.RowType flussFullRowType; private final int[] primaryKeyIndexes; private final Map tableOptions; + private final Duration lookupTimeout; + private final int executorThreads; + private final int maxConcurrency; - @Nullable private transient LakeTableLookuper lakeTableLookuper; - @Nullable private transient FixedSchemaDecoder lakeValueDecoder; - @Nullable private transient KeyEncoder lakePrimaryKeyEncoder; - @Nullable private transient KeyEncoder lakeBucketKeyEncoder; - @Nullable private transient BucketingFunction bucketingFunction; - - @Nullable - private transient org.apache.fluss.client.table.getter.PartitionGetter partitionGetter; + private LakeTableLookuper lakeTableLookuper; + private FixedSchemaDecoder lakeValueDecoder; + private KeyEncoder lakePrimaryKeyEncoder; + private KeyEncoder lakeBucketKeyEncoder; + private BucketingFunction bucketingFunction; + private PartitionGetter partitionGetter; + private ThreadPoolExecutor lookupExecutor; - private transient short lakeSchemaId; - private transient int numBuckets; + private short lakeSchemaId; + private int numBuckets; LakeLookupRuntime( + FlussLookupRuntime flussLookupRuntime, Configuration flussConfig, TablePath tablePath, org.apache.fluss.types.RowType flussFullRowType, int[] primaryKeyIndexes, - Map tableOptions) { - this.flussConfig = checkNotNull(flussConfig, "flussConfig must not be null."); - this.tablePath = checkNotNull(tablePath, "tablePath must not be null."); - this.flussFullRowType = - checkNotNull(flussFullRowType, "flussFullRowType must not be null."); - this.primaryKeyIndexes = - checkNotNull(primaryKeyIndexes, "primaryKeyIndexes must not be null."); - this.tableOptions = checkNotNull(tableOptions, "tableOptions must not be null."); + Map tableOptions, + Duration lookupTimeout, + int executorThreads, + int maxConcurrency) { + this.flussLookupRuntime = flussLookupRuntime; + this.flussConfig = flussConfig; + this.tablePath = tablePath; + this.flussFullRowType = flussFullRowType; + this.primaryKeyIndexes = primaryKeyIndexes; + this.tableOptions = tableOptions; + this.lookupTimeout = lookupTimeout; + this.executorThreads = executorThreads; + this.maxConcurrency = maxConcurrency; } - void open(TableInfo tableInfo) { - TableInfo resolvedTableInfo = checkNotNull(tableInfo, "tableInfo must not be null."); + @Override + public void open() { + TableInfo resolvedTableInfo = flussLookupRuntime.getTableInfo(); DataLakeFormat dataLakeFormat = validateAndGetDataLakeFormat(resolvedTableInfo); org.apache.fluss.types.RowType lookupRowType = flussFullRowType.project(primaryKeyIndexes); lakePrimaryKeyEncoder = @@ -101,9 +121,7 @@ void open(TableInfo tableInfo) { resolvedTableInfo.isDefaultBucketKey(), lakePrimaryKeyEncoder); bucketingFunction = BucketingFunction.of(dataLakeFormat); - partitionGetter = - new org.apache.fluss.client.table.getter.PartitionGetter( - lookupRowType, resolvedTableInfo.getPartitionKeys()); + partitionGetter = new PartitionGetter(lookupRowType, resolvedTableInfo.getPartitionKeys()); numBuckets = resolvedTableInfo.getNumBuckets(); lakeSchemaId = (short) resolvedTableInfo.getSchemaId(); lakeValueDecoder = @@ -111,56 +129,103 @@ void open(TableInfo tableInfo) { resolvedTableInfo.getTableConfig().getKvFormat(), resolvedTableInfo.getSchema()); lakeTableLookuper = createLakeTableLookuper(dataLakeFormat, resolvedTableInfo); + lookupExecutor = createLookupExecutor(); + } + + @Override + public CompletableFuture lookup(RowData normalizedKeyRow) { + final LakeTableLookuper lookuper = lakeTableLookuper; + final FixedSchemaDecoder valueDecoder = lakeValueDecoder; + final ThreadPoolExecutor executor = lookupExecutor; + final byte[] keyBytes; + final LakeTableLookuper.LookupContext lookupContext; + + try { + InternalRow lookupRow = new FlinkAsFlussRow(normalizedKeyRow); + keyBytes = encodePrimaryKey(lookupRow); + lookupContext = createLookupContext(lookupRow, keyBytes); + } catch (Throwable t) { + return FutureUtils.completedExceptionally(t); + } + + CompletableFuture future = new CompletableFuture<>(); + try { + executor.execute( + () -> { + try { + byte[] value = lookuper.lookup(keyBytes, lookupContext); + InternalRow row = + value == null + ? null + : valueDecoder.decode(MemorySegment.wrap(value)); + future.complete(new LookupResult(row)); + } catch (Throwable t) { + future.completeExceptionally( + new RuntimeException( + "Execution of lake fallback lookup failed: " + + t.getMessage(), + t)); + } + }); + } catch (RejectedExecutionException e) { + future.completeExceptionally( + new RuntimeException("Lake fallback lookup executor is overloaded.", e)); + } + + return FutureUtils.orTimeout( + future, + lookupTimeout.toMillis(), + TimeUnit.MILLISECONDS, + "Lake fallback lookup timed out after " + lookupTimeout); + } + + private byte[] encodePrimaryKey(InternalRow lookupRow) { + return lakePrimaryKeyEncoder.encodeKey(lookupRow); } - LakeLookupKey createLookupKey(RowData normalizedKeyRow) { - InternalRow lookupRow = new FlinkAsFlussRow(normalizedKeyRow); - KeyEncoder primaryKeyEncoder = - checkNotNull( - lakePrimaryKeyEncoder, "Lake primary-key encoder must be initialized."); - byte[] keyBytes = primaryKeyEncoder.encodeKey(lookupRow); + private LakeTableLookuper.LookupContext createLookupContext( + InternalRow lookupRow, byte[] keyBytes) { + final KeyEncoder primaryKeyEncoder = lakePrimaryKeyEncoder; byte[] bucketKeyBytes = lakeBucketKeyEncoder == primaryKeyEncoder ? keyBytes - : checkNotNull( - lakeBucketKeyEncoder, - "Lake bucket-key encoder must be initialized.") - .encodeKey(lookupRow); - int bucketId = - checkNotNull(bucketingFunction, "Bucketing function must be initialized.") - .bucketing(bucketKeyBytes, numBuckets); - ResolvedPartitionSpec partitionSpec = - checkNotNull(partitionGetter, "Partition getter must be initialized.") - .getResolvedPartitionSpec(lookupRow); + : lakeBucketKeyEncoder.encodeKey(lookupRow); + int bucketId = bucketingFunction.bucketing(bucketKeyBytes, numBuckets); + ResolvedPartitionSpec partitionSpec = partitionGetter.getResolvedPartitionSpec(lookupRow); LakeTableLookuper.LookupContext lookupContext = new LakeTableLookuper.LookupContext( partitionSpec, bucketId, lakeSchemaId, flussFullRowType); - return new LakeLookupKey(keyBytes, lookupContext); + return lookupContext; } - @Nullable - InternalRow lookup(LakeLookupKey lakeLookupKey) throws Exception { - byte[] value = - checkNotNull(lakeTableLookuper, "Lake table lookuper must be initialized.") - .lookup(lakeLookupKey.keyBytes, lakeLookupKey.lookupContext); - if (value == null) { - return null; + @Override + public void close() throws Exception { + if (lookupExecutor != null) { + lookupExecutor.shutdownNow(); } - return checkNotNull(lakeValueDecoder, "Lake value decoder must be initialized.") - .decode(MemorySegment.wrap(value)); - } - - void close() throws Exception { if (lakeTableLookuper != null) { lakeTableLookuper.close(); } } + private ThreadPoolExecutor createLookupExecutor() { + int queueCapacity = maxConcurrency - executorThreads; + BlockingQueue queue = + queueCapacity == 0 + ? new SynchronousQueue<>() + : new ArrayBlockingQueue<>(queueCapacity); + return new ThreadPoolExecutor( + executorThreads, + executorThreads, + 0L, + TimeUnit.MILLISECONDS, + queue, + new ExecutorThreadFactory("fluss-lake-fallback-lookup"), + new ThreadPoolExecutor.AbortPolicy()); + } + private DataLakeFormat validateAndGetDataLakeFormat(TableInfo tableInfo) { - DataLakeFormat dataLakeFormat = - checkNotNull( - tableInfo.getTableConfig().getDataLakeFormat().orElse(null), - "Data lake format must be configured for lake fallback lookup."); + DataLakeFormat dataLakeFormat = tableInfo.getTableConfig().getDataLakeFormat().orElse(null); if (dataLakeFormat != DataLakeFormat.PAIMON) { throw new TableException( "Hybrid lake lookup currently only supports Paimon, but table " @@ -190,19 +255,4 @@ private LakeTableLookuper createLakeTableLookuper( tableInfo.getTableConfig())), "Lake table lookuper must not be null."); } - - /** The encoded lake lookup key and its lake lookup context. */ - static final class LakeLookupKey { - private final byte[] keyBytes; - private final LakeTableLookuper.LookupContext lookupContext; - - private LakeLookupKey(byte[] keyBytes, LakeTableLookuper.LookupContext lookupContext) { - this.keyBytes = keyBytes; - this.lookupContext = lookupContext; - } - - ResolvedPartitionSpec getPartitionSpec() { - return lookupContext.partitionSpec(); - } - } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupRuntime.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupRuntime.java new file mode 100644 index 0000000000..1c92a42395 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/LookupRuntime.java @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.source.lookup; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.lookup.LookupResult; + +import org.apache.flink.table.data.RowData; + +import java.util.concurrent.CompletableFuture; + +/** Runtime abstraction for Flink lookup functions. */ +@Internal +public interface LookupRuntime extends AutoCloseable { + + /** Opens the runtime resources. */ + void open(); + + /** Looks up the given normalized key row. */ + CompletableFuture lookup(RowData keyRow); +} From 3c47ec5919e001c8d05a0081bd2324616ee3c337 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Fri, 7 Aug 2026 16:05:29 +0800 Subject: [PATCH 5/6] fix --- .../lookup/HybridLakeAsyncLookupFunction.java | 21 ++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java index 427fb567af..6989a7f441 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunction.java @@ -48,6 +48,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.function.Supplier; import java.util.stream.IntStream; /** @@ -186,8 +187,15 @@ public CompletableFuture> asyncLookup(RowData keyRow) { if (partitionExists == null) { partitionExists = getOrRefreshPartitionExistence(partitionSpec); } + if (partitionExists) { - return lookupFlussAsync(normalizedKeyRow, partitionSpec, remainingFilter); + return lookupFlussAsync( + normalizedKeyRow, + remainingFilter, + () -> { + markPartitionInvalid(partitionSpec); + return lookupLakeAsync(normalizedKeyRow, remainingFilter); + }); } else { return lookupLakeAsync(normalizedKeyRow, remainingFilter); } @@ -198,8 +206,8 @@ public CompletableFuture> asyncLookup(RowData keyRow) { private CompletableFuture> lookupFlussAsync( RowData normalizedKeyRow, - PartitionSpec partitionSpec, - @Nullable LookupNormalizer.RemainingFilter remainingFilter) { + @Nullable LookupNormalizer.RemainingFilter remainingFilter, + Supplier>> lakefallbackLookupFunc) { CompletableFuture> future = new CompletableFuture<>(); try { flussLookupRuntime @@ -211,8 +219,8 @@ private CompletableFuture> lookupFlussAsync( if (ExceptionUtils.findThrowable( throwable, PartitionNotExistException.class) .isPresent()) { - markPartitionInvalid(partitionSpec); - lookupLakeAsync(normalizedKeyRow, remainingFilter) + lakefallbackLookupFunc + .get() .whenComplete( (rows, error) -> FutureUtils.doForward( @@ -244,8 +252,7 @@ private CompletableFuture> lookupFlussAsync( }); } catch (Throwable t) { if (ExceptionUtils.findThrowable(t, PartitionNotExistException.class).isPresent()) { - markPartitionInvalid(partitionSpec); - return lookupLakeAsync(normalizedKeyRow, remainingFilter); + return lakefallbackLookupFunc.get(); } else { future.completeExceptionally(t); } From 02246458de22035f5d02ea48ed2188d162760c78 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Fri, 7 Aug 2026 16:36:29 +0800 Subject: [PATCH 6/6] add tests --- .../HybridLakeAsyncLookupFunctionTest.java | 227 ++++++++++++++++++ .../TestingPaimonLakeStoragePlugin.java | 72 ++++++ .../src/test/resources/log4j2-test.properties | 2 +- 3 files changed, 300 insertions(+), 1 deletion(-) create mode 100644 fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunctionTest.java diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunctionTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunctionTest.java new file mode 100644 index 0000000000..f2f8e1e184 --- /dev/null +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/lookup/HybridLakeAsyncLookupFunctionTest.java @@ -0,0 +1,227 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.flink.source.lookup; + +import org.apache.fluss.client.table.Table; +import org.apache.fluss.client.table.writer.UpsertWriter; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.flink.tiering.source.TieringTestBase; +import org.apache.fluss.flink.utils.FlinkConversions; +import org.apache.fluss.lake.values.TestingPaimonLakeStoragePlugin; +import org.apache.fluss.metadata.DataLakeFormat; +import org.apache.fluss.metadata.PartitionSpec; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.types.DataTypes; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.fluss.flink.source.lookup.LookupNormalizer.createPrimaryKeyLookupNormalizer; +import static org.apache.fluss.testutils.DataTestUtils.row; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link HybridLakeAsyncLookupFunction}. */ +class HybridLakeAsyncLookupFunctionTest extends TieringTestBase { + + private static final String EXISTING_PARTITION = "2026"; + private static final String MISSING_PARTITION = "1900"; + private static final TablePath TABLE_PATH = + TablePath.of(DEFAULT_DB, "hybrid-lake-lookup-table"); + private static final Schema TABLE_SCHEMA = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("date", DataTypes.STRING()) + .primaryKey("id", "date") + .build(); + private static final TableDescriptor TABLE_DESCRIPTOR = + TableDescriptor.builder() + .schema(TABLE_SCHEMA) + .distributedBy(1) + .partitionedBy("date") + .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true) + .property(ConfigOptions.TABLE_DATALAKE_FORMAT, DataLakeFormat.PAIMON) + .build(); + + private HybridLakeAsyncLookupFunction lookupFunction; + + @BeforeEach + void setUp() throws Exception { + TestingPaimonLakeStoragePlugin.resetLookupFunction(); + admin.createTable(TABLE_PATH, TABLE_DESCRIPTOR, true).get(); + admin.createPartition(TABLE_PATH, partitionSpec(EXISTING_PARTITION), true).get(); + Map partitionIds = + FLUSS_CLUSTER_EXTENSION.waitUntilPartitionAllReady(TABLE_PATH, 1); + long tableId = admin.getTableInfo(TABLE_PATH).get().getTableId(); + FLUSS_CLUSTER_EXTENSION.waitUntilTablePartitionReady( + tableId, partitionIds.get(EXISTING_PARTITION)); + } + + @AfterEach + void tearDown() throws Exception { + if (lookupFunction != null) { + lookupFunction.close(); + lookupFunction = null; + } + TestingPaimonLakeStoragePlugin.resetLookupFunction(); + } + + @Test + void testLookupExistingFlussPartitionWithoutLakeFallback() throws Exception { + AtomicInteger lakeLookupCount = new AtomicInteger(); + TestingPaimonLakeStoragePlugin.setLookupFunction( + (key, context) -> { + lakeLookupCount.incrementAndGet(); + return row(1, "lake", EXISTING_PARTITION); + }); + writeRow(row(1, "fluss", EXISTING_PARTITION)); + openLookupFunction(Duration.ofSeconds(5)); + + assertThat(lookup(1, EXISTING_PARTITION)) + .singleElement() + .extracting(RowData::toString) + .isEqualTo("+I(1,fluss,2026)"); + assertThat(lookup(2, EXISTING_PARTITION)).isEmpty(); + assertThat(lakeLookupCount).hasValue(0); + } + + @Test + void testFallbackToLakeForMissingPartition() throws Exception { + AtomicInteger lakeLookupCount = new AtomicInteger(); + TestingPaimonLakeStoragePlugin.setLookupFunction( + (key, context) -> { + lakeLookupCount.incrementAndGet(); + assertThat(context.partitionSpec().toPartitionSpec()) + .isEqualTo(partitionSpec(MISSING_PARTITION)); + return row(3, "lake", MISSING_PARTITION); + }); + openLookupFunction(Duration.ofSeconds(5)); + + assertThat(lookup(3, MISSING_PARTITION)) + .singleElement() + .extracting(RowData::toString) + .isEqualTo("+I(3,lake,1900)"); + assertThat(lakeLookupCount).hasValue(1); + } + + @Test + void testEmptyLakeLookupResult() throws Exception { + openLookupFunction(Duration.ofSeconds(5)); + + assertThat(lookup(4, MISSING_PARTITION)).isEmpty(); + } + + @Test + void testLakeLookupFailureIsPropagated() { + TestingPaimonLakeStoragePlugin.setLookupFunction( + (key, context) -> { + throw new IOException("lake lookup failure"); + }); + openLookupFunction(Duration.ofSeconds(5)); + + assertThatThrownBy(() -> lookup(5, MISSING_PARTITION)) + .isInstanceOf(ExecutionException.class) + .hasRootCauseInstanceOf(IOException.class) + .hasRootCauseMessage("lake lookup failure"); + } + + @Test + void testLakeLookupTimeoutIsPropagated() throws Exception { + CountDownLatch lookupStarted = new CountDownLatch(1); + CountDownLatch releaseLookup = new CountDownLatch(1); + TestingPaimonLakeStoragePlugin.setLookupFunction( + (key, context) -> { + lookupStarted.countDown(); + releaseLookup.await(); + return row(6, "lake", MISSING_PARTITION); + }); + openLookupFunction(Duration.ofMillis(100)); + + CompletableFuture> future = + lookupFunction.asyncLookup(lookupKey(6, MISSING_PARTITION)); + assertThat(lookupStarted.await(10, TimeUnit.SECONDS)).isTrue(); + try { + assertThatThrownBy(() -> future.get(10, TimeUnit.SECONDS)) + .isInstanceOf(ExecutionException.class) + .hasRootCauseInstanceOf(TimeoutException.class); + } finally { + releaseLookup.countDown(); + } + } + + private void openLookupFunction(Duration timeout) { + RowType flinkRowType = FlinkConversions.toFlinkRowType(TABLE_SCHEMA.getRowType()); + int[] primaryKeyIndexes = TABLE_SCHEMA.getPrimaryKeyIndexes(); + lookupFunction = + new HybridLakeAsyncLookupFunction( + clientConf, + TABLE_PATH, + flinkRowType, + primaryKeyIndexes, + createPrimaryKeyLookupNormalizer(primaryKeyIndexes, flinkRowType), + null, + Collections.singletonMap( + ConfigOptions.TABLE_DATALAKE_FORMAT.key(), + DataLakeFormat.PAIMON.toString()), + timeout, + 1, + 2); + lookupFunction.open(null); + } + + private Collection lookup(int id, String partition) throws Exception { + return lookupFunction.asyncLookup(lookupKey(id, partition)).get(10, TimeUnit.SECONDS); + } + + private static void writeRow(InternalRow row) throws Exception { + try (Table table = conn.getTable(TABLE_PATH)) { + UpsertWriter writer = table.newUpsert().createWriter(); + writer.upsert(row); + writer.flush(); + } + } + + private static GenericRowData lookupKey(int id, String partition) { + return GenericRowData.of(id, StringData.fromString(partition)); + } + + private static PartitionSpec partitionSpec(String partition) { + return new PartitionSpec(Collections.singletonMap("date", partition)); + } +} diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java index a8844f24db..97849e5fd4 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/TestingPaimonLakeStoragePlugin.java @@ -24,6 +24,7 @@ import org.apache.fluss.lake.lakestorage.LakeCatalog; import org.apache.fluss.lake.lakestorage.LakeStorage; import org.apache.fluss.lake.lakestorage.LakeStoragePlugin; +import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.serializer.SimpleVersionedSerializer; import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.source.LakeSplit; @@ -35,14 +36,37 @@ import org.apache.fluss.metadata.TableDescriptor; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.predicate.Predicate; +import org.apache.fluss.row.BinaryRow; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.row.encode.ValueEncoder; + +import javax.annotation.Nullable; import java.io.IOException; import java.util.Collections; import java.util.List; +import static org.apache.fluss.utils.Preconditions.checkNotNull; + /** Test-only Paimon lake storage plugin used to construct Flink table sources. */ public class TestingPaimonLakeStoragePlugin implements LakeStoragePlugin { + private static final LookupFunction EMPTY_LOOKUP = (key, context) -> null; + + private static volatile LookupFunction lookupFunction = EMPTY_LOOKUP; + + /** Configures the point lookup behavior used by newly-created and existing test lookupers. */ + public static void setLookupFunction(LookupFunction lookupFunction) { + TestingPaimonLakeStoragePlugin.lookupFunction = + checkNotNull(lookupFunction, "lookupFunction must not be null."); + } + + /** Resets point lookups to return no row. */ + public static void resetLookupFunction() { + lookupFunction = EMPTY_LOOKUP; + } + @Override public String identifier() { return DataLakeFormat.PAIMON.toString(); @@ -68,6 +92,45 @@ public LakeCatalog createLakeCatalog() { public LakeSource createLakeSource(TablePath tablePath) { return new TestingPaimonLakeSource(); } + + @Override + public LakeTableLookuper createLakeTableLookuper( + TablePath tablePath, LookuperContext context) { + return new TestingLakeTableLookuper(context); + } + } + + private static class TestingLakeTableLookuper implements LakeTableLookuper { + + private final LakeStorage.LookuperContext lookuperContext; + + private TestingLakeTableLookuper(LakeStorage.LookuperContext lookuperContext) { + this.lookuperContext = lookuperContext; + } + + @Override + public @Nullable byte[] lookup(byte[] key, LookupContext context) throws Exception { + InternalRow row = lookupFunction.lookup(key, context); + if (row == null) { + return null; + } + + try (RowEncoder rowEncoder = + RowEncoder.create( + lookuperContext.tableConfig().getKvFormat(), context.valueRowType())) { + InternalRow.FieldGetter[] fieldGetters = + InternalRow.createFieldGetters(context.valueRowType()); + rowEncoder.startNewRow(); + for (int i = 0; i < fieldGetters.length; i++) { + rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(row)); + } + BinaryRow binaryRow = rowEncoder.finishRow(); + return ValueEncoder.encodeValue(context.schemaId(), binaryRow); + } + } + + @Override + public void close() {} } private static class TestingPaimonLakeCatalog implements LakeCatalog { @@ -108,4 +171,13 @@ public SimpleVersionedSerializer getSplitSerializer() { throw new UnsupportedOperationException("Not implemented."); } } + + /** Test callback for a lake point lookup. */ + @FunctionalInterface + public interface LookupFunction { + + /** Returns the row for the encoded key, or null when the key is absent. */ + @Nullable + InternalRow lookup(byte[] key, LakeTableLookuper.LookupContext context) throws Exception; + } } diff --git a/fluss-flink/fluss-flink-common/src/test/resources/log4j2-test.properties b/fluss-flink/fluss-flink-common/src/test/resources/log4j2-test.properties index 38f3ade868..b1e9100b98 100644 --- a/fluss-flink/fluss-flink-common/src/test/resources/log4j2-test.properties +++ b/fluss-flink/fluss-flink-common/src/test/resources/log4j2-test.properties @@ -29,4 +29,4 @@ appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n # suppress the duplicated logger extension logger.flink.name = org.apache.flink.util.TestLoggerExtension -logger.flink.level = OFF \ No newline at end of file +logger.flink.level = OFF