[flink] add support of hybrid lookup with lake

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -284,6 +284,21 @@ public class MetricNames {
     public static final String SCANNER_REMOTE_FETCH_RATE = "remoteFetchRequestsPerSecond";
     public static final String SCANNER_REMOTE_FETCH_ERROR_RATE = "remoteFetchErrorPerSecond";
 
+    // for flink hybrid lake lookup
+    public static final String LOOKUP_HOT_FLUSS_HITS_TOTAL = "lookupHotFlussHitsTotal";
+    public static final String LOOKUP_HOT_FLUSS_MISSES_TOTAL = "lookupHotFlussMissesTotal";
+    public static final String LOOKUP_COLD_FLUSS_HITS_TOTAL = "lookupColdFlussHitsTotal";
+    public static final String LOOKUP_COLD_FLUSS_MISSES_TOTAL = "lookupColdFlussMissesTotal";
+    public static final String LAKE_FALLBACK_REQUESTS_TOTAL = "lakeFallbackRequestsTotal";
+    public static final String LAKE_FALLBACK_HITS_TOTAL = "lakeFallbackHitsTotal";
+    public static final String LAKE_FALLBACK_MISSES_TOTAL = "lakeFallbackMissesTotal";
+    public static final String LAKE_FALLBACK_FAILURES_TOTAL = "lakeFallbackFailuresTotal";
+    public static final String LAKE_FALLBACK_TIMEOUTS_TOTAL = "lakeFallbackTimeoutsTotal";
+    public static final String LAKE_FALLBACK_REJECTED_TOTAL = "lakeFallbackRejectedTotal";
+    public static final String LAKE_FALLBACK_LATENCY_MS = "lakeFallbackLatencyMs";
+    public static final String LAKE_FALLBACK_PENDING_COUNT = "lakeFallbackPendingCount";
+    public static final String LAKE_FALLBACK_QUEUE_SIZE = "lakeFallbackQueueSize";
+
     // for netty
     public static final String NETTY_USED_DIRECT_MEMORY = "usedDirectMemory";
     public static final String NETTY_NUM_DIRECT_ARENAS = "numDirectArenas";

fluss-flink/fluss-flink-1.20/src/test/java/org/apache/fluss/flink/source/Flink120HybridLakeLookupITCase.java

@@ -0,0 +1,21 @@
+/*
+ * 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;
+
+/** IT case for hybrid lake lookup in Flink 1.20. */
+public class Flink120HybridLakeLookupITCase extends HybridLakeLookupITCase {}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java

@@ -115,6 +115,47 @@ public class FlinkConnectorOptions {
                                     + "with the lookup key values. This feature cannot be used with PREFIX_LOOKUP type. "
                                     + "Default is false.");
 
+    public static final ConfigOption<Boolean> LOOKUP_LAKE_FALLBACK_ENABLED =
+            ConfigOptions.key("lookup.lake-fallback.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to fall back to the lake table when asynchronous lookup misses in Fluss.");
+
+    public static final ConfigOption<Duration> LOOKUP_HOT_WINDOW =
+            ConfigOptions.key("lookup.hot-window")
+                    .durationType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The hot data window for lake fallback lookup. Lookup keys inside this window only query Fluss.");
+
+    public static final ConfigOption<String> LOOKUP_TIME_ZONE =
+            ConfigOptions.key("lookup.time-zone")
+                    .stringType()
+                    .defaultValue(java.time.ZoneId.systemDefault().getId())
+                    .withDescription(
+                            "The time zone used to interpret hour partition values for lake fallback lookup.");
+
+    public static final ConfigOption<Duration> LOOKUP_LAKE_FALLBACK_TIMEOUT =
+            ConfigOptions.key("lookup.lake-fallback.timeout")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(30))
+                    .withDescription("The timeout for a single lake fallback lookup.");
+
+    public static final ConfigOption<Integer> LOOKUP_LAKE_FALLBACK_EXECUTOR_THREADS =
+            ConfigOptions.key("lookup.lake-fallback.executor-threads")
+                    .intType()
+                    .defaultValue(4)
+                    .withDescription(
+                            "The number of worker threads used for blocking lake fallback lookups.");
+
+    public static final ConfigOption<Integer> LOOKUP_LAKE_FALLBACK_MAX_CONCURRENCY =
+            ConfigOptions.key("lookup.lake-fallback.max-concurrency")
+                    .intType()
+                    .defaultValue(1024)
+                    .withDescription(
+                            "The maximum number of active and queued lake fallback lookups per lookup function instance.");
+
     // --------------------------------------------------------------------------------------------
     // Scan specific options
     // --------------------------------------------------------------------------------------------

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java

@@ -51,6 +51,7 @@
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.flink.table.types.logical.RowType;
 
+import java.time.Duration;
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -148,6 +149,8 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                         .toMillis();
         int splitAssignmentBatchSize =
                 tableOptions.get(FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE);
+        Duration lookupHotWindow =
+                tableOptions.getOptional(FlinkConnectorOptions.LOOKUP_HOT_WINDOW).orElse(null);
 
         LeaseContext leaseContext = LeaseContext.fromConf(tableOptions);
         return new FlinkTableSource(
@@ -163,6 +166,12 @@ 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),
+                lookupHotWindow,
+                tableOptions.get(FlinkConnectorOptions.LOOKUP_TIME_ZONE),
+                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 +251,12 @@ public Set<ConfigOption<?>> optionalOptions() {
                                 FlinkConnectorOptions.SCAN_KV_SNAPSHOT_LEASE_DURATION,
                                 FlinkConnectorOptions.LOOKUP_ASYNC,
                                 FlinkConnectorOptions.LOOKUP_INSERT_IF_NOT_EXISTS,
+                                FlinkConnectorOptions.LOOKUP_LAKE_FALLBACK_ENABLED,
+                                FlinkConnectorOptions.LOOKUP_HOT_WINDOW,
+                                FlinkConnectorOptions.LOOKUP_TIME_ZONE,
+                                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,