(iceberg) branch main updated: Flink: Backport Support Lookup Join using full in-memory lookup cache (#18766)

[mxm.apache.org via commits]

This is an automated email from the ASF dual-hosted git repository.

mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 66a1417a76 Flink: Backport Support Lookup Join using full in-memory 
lookup cache (#18766)
66a1417a76 is described below

commit 66a1417a7674d677aee6c6d24cc6486ec33b2192
Author: GuoYu <511955...@qq.com>
AuthorDate: Wed Nov 4 18:49:55 2026 +0800

    Flink: Backport Support Lookup Join using full in-memory lookup cache 
(#18766)
---
 .../iceberg/flink/FlinkDynamicTableFactory.java    |   2 +
 .../iceberg/flink/source/IcebergTableSource.java   |  60 +++-
 .../lookup/IcebergFullCachingLookupFunction.java   | 223 +++++++++++++
 .../flink/source/lookup/IcebergLookupOptions.java  |  41 +++
 .../flink/source/lookup/IcebergLookupReader.java   | 124 +++++++
 .../flink/source/lookup/InMemoryLookupCache.java   |  69 ++++
 .../iceberg/flink/source/TestFlinkSourceSql.java   |   2 +-
 .../flink/source/TestIcebergLookupJoinSql.java     | 369 +++++++++++++++++++++
 .../iceberg/flink/source/TestIcebergSourceSql.java |   2 +-
 .../iceberg/flink/source/TestSourceSqlBase.java}   |  54 +--
 .../apache/iceberg/flink/source/TestSqlBase.java   |  99 +-----
 .../flink/source/lookup/TestIcebergLookup.java     | 318 ++++++++++++++++++
 .../iceberg/flink/FlinkDynamicTableFactory.java    |   2 +
 .../iceberg/flink/source/IcebergTableSource.java   |  60 +++-
 .../lookup/IcebergFullCachingLookupFunction.java   | 223 +++++++++++++
 .../flink/source/lookup/IcebergLookupOptions.java  |  41 +++
 .../flink/source/lookup/IcebergLookupReader.java   | 124 +++++++
 .../flink/source/lookup/InMemoryLookupCache.java   |  69 ++++
 .../iceberg/flink/source/TestFlinkSourceSql.java   |   2 +-
 .../flink/source/TestIcebergLookupJoinSql.java     | 369 +++++++++++++++++++++
 .../iceberg/flink/source/TestIcebergSourceSql.java |   2 +-
 .../{TestSqlBase.java => TestSourceSqlBase.java}   |  54 +--
 .../apache/iceberg/flink/source/TestSqlBase.java   |  97 +-----
 .../flink/source/lookup/TestIcebergLookup.java     | 318 ++++++++++++++++++
 gradle.properties                                  |   2 +-
 25 files changed, 2419 insertions(+), 307 deletions(-)

diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index 6306ee7a0a..0c7244dc23 100644
--- 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
@@ -39,6 +39,7 @@ import 
org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.flink.source.IcebergTableSource;
+import org.apache.iceberg.flink.source.lookup.IcebergLookupOptions;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
@@ -132,6 +133,7 @@ public class FlinkDynamicTableFactory
     options.add(FlinkCreateTableOptions.CATALOG_TABLE);
     options.add(FlinkCreateTableOptions.USE_DYNAMIC_ICEBERG_SINK);
     options.add(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+    options.add(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD);
     return options;
   }
 
diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
index c65a423489..7cd93f53e0 100644
--- 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
@@ -34,20 +34,28 @@ import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.ProviderContext;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
 import 
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
 import 
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import 
org.apache.flink.table.connector.source.abilities.SupportsSourceWatermark;
+import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupOptions;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.expressions.ResolvedExpression;
 import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.LookupFunction;
+import org.apache.flink.table.types.logical.RowType;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkConfParser;
 import org.apache.iceberg.flink.FlinkConfigOptions;
 import org.apache.iceberg.flink.FlinkFilters;
 import org.apache.iceberg.flink.FlinkReadOptions;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.source.assigner.SplitAssignerType;
+import org.apache.iceberg.flink.source.lookup.IcebergFullCachingLookupFunction;
+import org.apache.iceberg.flink.source.lookup.IcebergLookupOptions;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -60,7 +68,8 @@ public class IcebergTableSource
         SupportsProjectionPushDown,
         SupportsFilterPushDown,
         SupportsLimitPushDown,
-        SupportsSourceWatermark {
+        SupportsSourceWatermark,
+        LookupTableSource {
 
   private int[] projectedFields;
   private Long limit;
@@ -71,6 +80,7 @@ public class IcebergTableSource
   private final Map<String, String> properties;
   private final boolean isLimitPushDown;
   private final ReadableConfig readableConfig;
+  private final FlinkConfParser flinkConfParser;
 
   private IcebergTableSource(IcebergTableSource toCopy) {
     this.loader = toCopy.loader;
@@ -81,6 +91,7 @@ public class IcebergTableSource
     this.limit = toCopy.limit;
     this.filters = toCopy.filters;
     this.readableConfig = toCopy.readableConfig;
+    this.flinkConfParser = toCopy.flinkConfParser;
   }
 
   public IcebergTableSource(
@@ -108,6 +119,7 @@ public class IcebergTableSource
     this.limit = limit;
     this.filters = filters;
     this.readableConfig = readableConfig;
+    this.flinkConfParser = new FlinkConfParser(properties, readableConfig);
   }
 
   @Override
@@ -236,4 +248,50 @@ public class IcebergTableSource
   public String asSummaryString() {
     return "Iceberg table source";
   }
+
+  @Override
+  public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) 
{
+    int[][] lookupKeys = context.getKeys();
+    int[] lookupKeyIndexes = new int[lookupKeys.length];
+    for (int i = 0; i < lookupKeys.length; i++) {
+      Preconditions.checkArgument(
+          lookupKeys[i].length == 1, "Iceberg lookup source doesn't support 
nested lookup key.");
+      lookupKeyIndexes[i] = lookupKeys[i][0];
+    }
+
+    ResolvedSchema projected = getProjectedSchema();
+    RowType projectedRowType = (RowType) 
projected.toPhysicalRowDataType().getLogicalType();
+
+    LookupOptions.LookupCacheType requestedCacheType =
+        flinkConfParser
+            .enumConfParser(LookupOptions.LookupCacheType.class)
+            .option(LookupOptions.CACHE_TYPE.key())
+            .parseOptional();
+    Preconditions.checkArgument(
+        requestedCacheType == null || requestedCacheType == 
LookupOptions.LookupCacheType.FULL,
+        "Iceberg lookup join only supports %s=FULL, but it is set to '%s'. 
NONE and PARTIAL are "
+            + "not supported, because an Iceberg table cannot be 
point-looked-up.",
+        LookupOptions.CACHE_TYPE.key(),
+        requestedCacheType);
+
+    boolean eagerLoad =
+        flinkConfParser
+            .booleanConf()
+            .option(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD.key())
+            
.defaultValue(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD.defaultValue())
+            .parse();
+
+    boolean caseSensitive =
+        flinkConfParser
+            .booleanConf()
+            .option(FlinkReadOptions.CASE_SENSITIVE)
+            .flinkConfig(FlinkReadOptions.CASE_SENSITIVE_OPTION)
+            
.defaultValue(FlinkReadOptions.CASE_SENSITIVE_OPTION.defaultValue())
+            .parse();
+
+    LookupFunction lookupFn =
+        new IcebergFullCachingLookupFunction(
+            loader, projectedRowType, lookupKeyIndexes, filters, 
caseSensitive, eagerLoad);
+    return LookupFunctionProvider.of(lookupFn);
+  }
 }
diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
new file mode 100644
index 0000000000..d7e78913e3
--- /dev/null
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
@@ -0,0 +1,223 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.LookupFunction;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkRowData;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.io.CloseableIterable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A lookup function that serves every lookup from an in-memory cache of the 
rows of an Iceberg
+ * table, after projection and pushed-down filters are applied.
+ *
+ * <p>The cache is loaded from the current snapshot either when the function 
is opened or on the
+ * first lookup, depending on {@link 
IcebergLookupOptions#FULL_CACHE_EAGER_LOAD}. It is never
+ * refreshed, so later changes to the table are not visible.
+ */
+@Internal
+public class IcebergFullCachingLookupFunction extends LookupFunction {
+
+  private static final Logger LOG = 
LoggerFactory.getLogger(IcebergFullCachingLookupFunction.class);
+
+  private static final String METRIC_GROUP = "cache";
+  private static final long UNINITIALIZED = -1L;
+
+  private final TableLoader tableLoader;
+  private final RowType projectedRowType;
+  private final int[] lookupKeyIndexes;
+  private final List<Expression> pushedFilters;
+  private final boolean caseSensitive;
+  private final boolean eagerLoad;
+
+  private transient Table table;
+  private transient IcebergLookupReader reader;
+  private transient RowDataSerializer rowSerializer;
+  private transient RowDataSerializer keySerializer;
+  private transient RowData.FieldGetter[] keyGetters;
+  private transient InMemoryLookupCache cache;
+
+  private transient Counter missCounter;
+  private transient volatile long currentSnapshotId;
+  private transient volatile long cachedRecords;
+  private transient volatile long latestLoadTimeMs;
+
+  public IcebergFullCachingLookupFunction(
+      TableLoader tableLoader,
+      RowType projectedRowType,
+      int[] lookupKeyIndexes,
+      List<Expression> pushedFilters,
+      boolean caseSensitive,
+      boolean eagerLoad) {
+    this.tableLoader = tableLoader;
+    this.projectedRowType = projectedRowType;
+    this.lookupKeyIndexes = lookupKeyIndexes;
+    this.pushedFilters = pushedFilters;
+    this.caseSensitive = caseSensitive;
+    this.eagerLoad = eagerLoad;
+  }
+
+  @Override
+  public void open(FunctionContext context) throws Exception {
+    super.open(context);
+    LOG.info(
+        "IcebergFullCachingLookupFunction opening, projected fields={}, 
lookupKeyIndexes={}",
+        projectedRowType.getFieldNames(),
+        Arrays.toString(lookupKeyIndexes));
+
+    MetricGroup metricGroup = context.getMetricGroup().addGroup(METRIC_GROUP);
+    this.missCounter = metricGroup.counter("missCount");
+    this.currentSnapshotId = IcebergLookupReader.CURRENT_SNAPSHOT;
+    this.cachedRecords = 0L;
+    this.latestLoadTimeMs = UNINITIALIZED;
+    metricGroup.gauge("snapshotId", () -> currentSnapshotId);
+    metricGroup.gauge("numCachedRecords", () -> cachedRecords);
+    metricGroup.gauge("latestLoadTime", () -> latestLoadTimeMs);
+
+    tableLoader.open();
+    this.table = tableLoader.loadTable();
+
+    Schema icebergProjection =
+        FlinkSchemaUtil.convert(table.schema(), 
FlinkSchemaUtil.toResolvedSchema(projectedRowType));
+
+    createAccessors();
+
+    String nameMapping = 
table.properties().get(TableProperties.DEFAULT_NAME_MAPPING);
+    this.reader =
+        new IcebergLookupReader(
+            table, icebergProjection, pushedFilters, caseSensitive, 
nameMapping);
+
+    if (eagerLoad) {
+      loadCache();
+    }
+  }
+
+  /**
+   * Returns the cached rows for the key. The returned collection and its rows 
are shared with the
+   * cache, so they must not be modified.
+   */
+  @Override
+  public Collection<RowData> lookup(RowData keyRow) throws IOException {
+    if (cache == null) {
+      loadCache();
+    }
+
+    List<RowData> hit = cache.get(keySerializer.toBinaryRow(keyRow));
+    if (hit == null) {
+      missCounter.inc();
+      return Collections.emptyList();
+    }
+
+    return hit;
+  }
+
+  @Override
+  public void close() throws Exception {
+    try {
+      tableLoader.close();
+    } finally {
+      if (cache != null) {
+        cache.close();
+      }
+
+      super.close();
+    }
+  }
+
+  private void loadCache() throws IOException {
+    table.refresh();
+    Snapshot snapshot = table.currentSnapshot();
+    long snapshotId =
+        snapshot == null ? IcebergLookupReader.CURRENT_SNAPSHOT : 
snapshot.snapshotId();
+
+    LOG.info(
+        "IcebergFullCachingLookupFunction loading started, snapshot={}, 
committedAt={}, projected fields={}, pushedFilters={}",
+        snapshot == null ? "none" : snapshotId,
+        snapshot == null ? "n/a" : snapshot.timestampMillis(),
+        projectedRowType.getFieldNames(),
+        pushedFilters);
+
+    InMemoryLookupCache loaded = new InMemoryLookupCache();
+    long rowCnt = 0L;
+    long start = System.currentTimeMillis();
+    try (CloseableIterable<RowData> rows = reader.read(snapshotId)) {
+      for (RowData row : rows) {
+        loaded.add(extractKey(row), rowSerializer.toBinaryRow(row).copy());
+        rowCnt++;
+      }
+    } finally {
+      this.latestLoadTimeMs = System.currentTimeMillis() - start;
+    }
+
+    this.cache = loaded;
+    this.currentSnapshotId = snapshotId;
+    this.cachedRecords = rowCnt;
+
+    LOG.info(
+        "IcebergFullCachingLookupFunction loading finished, snapshot={}, 
rows={}, cost={} ms",
+        snapshot == null ? "none" : snapshotId,
+        rowCnt,
+        latestLoadTimeMs);
+  }
+
+  private BinaryRowData extractKey(RowData row) {
+    GenericRowData key = new GenericRowData(keyGetters.length);
+    for (int i = 0; i < keyGetters.length; i++) {
+      key.setField(i, keyGetters[i].getFieldOrNull(row));
+    }
+
+    return keySerializer.toBinaryRow(key).copy();
+  }
+
+  private void createAccessors() {
+    LogicalType[] keyTypes = new LogicalType[lookupKeyIndexes.length];
+    this.keyGetters = new RowData.FieldGetter[lookupKeyIndexes.length];
+    for (int i = 0; i < lookupKeyIndexes.length; i++) {
+      int projectedIndex = lookupKeyIndexes[i];
+      keyTypes[i] = projectedRowType.getTypeAt(projectedIndex);
+      this.keyGetters[i] = FlinkRowData.createFieldGetter(keyTypes[i], 
projectedIndex);
+    }
+
+    this.rowSerializer = new RowDataSerializer(projectedRowType);
+    this.keySerializer = new RowDataSerializer(keyTypes);
+  }
+}
diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
new file mode 100644
index 0000000000..355fd5387f
--- /dev/null
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
@@ -0,0 +1,41 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/** Lookup join options for the Iceberg table source. */
+@Internal
+public class IcebergLookupOptions {
+
+  public static final ConfigOption<Boolean> FULL_CACHE_EAGER_LOAD =
+      ConfigOptions.key("lookup.full-cache.eager-load")
+          .booleanType()
+          .defaultValue(true)
+          .withDescription(
+              "Whether to load the full lookup cache when the lookup function 
is opened, instead "
+                  + "of on the first lookup. Eager loading narrows the window 
in which the subtasks "
+                  + "of the join can end up on different snapshots of the 
dimension table, and "
+                  + "fails the job at startup if the table cannot be read, at 
the cost of a longer "
+                  + "deployment.");
+
+  private IcebergLookupOptions() {}
+}
diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
new file mode 100644
index 0000000000..d669178557
--- /dev/null
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
@@ -0,0 +1,124 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import java.io.IOException;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.flink.source.DataIterator;
+import org.apache.iceberg.flink.source.FileScanTaskReader;
+import org.apache.iceberg.flink.source.RowDataFileScanTaskReader;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.util.SnapshotUtil;
+
+/** Reads all rows of an Iceberg table to build a lookup cache. */
+@Internal
+class IcebergLookupReader {
+
+  static final long CURRENT_SNAPSHOT = -1L;
+
+  private final Table table;
+  private final Schema projectedSchema;
+  private final List<Expression> baseFilters;
+  private final Expression filter;
+  private final boolean caseSensitive;
+  private final String nameMapping;
+
+  IcebergLookupReader(
+      Table table,
+      Schema projectedSchema,
+      List<Expression> baseFilters,
+      boolean caseSensitive,
+      @Nullable String nameMapping) {
+    Preconditions.checkNotNull(baseFilters, "Base filters should not be null");
+
+    this.table = table;
+    this.projectedSchema = projectedSchema;
+    this.baseFilters = baseFilters;
+    this.caseSensitive = caseSensitive;
+    this.nameMapping = nameMapping;
+
+    Expression combinedFilter = Expressions.alwaysTrue();
+    for (Expression baseFilter : this.baseFilters) {
+      combinedFilter = Expressions.and(combinedFilter, baseFilter);
+    }
+
+    this.filter = combinedFilter;
+  }
+
+  CloseableIterable<RowData> read(long snapshotId) {
+    Schema tableSchema =
+        snapshotId == CURRENT_SNAPSHOT ? table.schema() : 
SnapshotUtil.schemaFor(table, snapshotId);
+
+    RowDataFileScanTaskReader fileReader =
+        new RowDataFileScanTaskReader(
+            tableSchema, projectedSchema, nameMapping, caseSensitive, 
baseFilters);
+
+    TableScan scan =
+        
table.newScan().caseSensitive(caseSensitive).project(projectedSchema).filter(filter);
+    if (snapshotId != CURRENT_SNAPSHOT) {
+      scan = scan.useSnapshot(snapshotId);
+    }
+
+    return new LookupIterable(scan.planTasks(), fileReader);
+  }
+
+  private class LookupIterable extends CloseableGroup implements 
CloseableIterable<RowData> {
+    private final CloseableIterable<CombinedScanTask> tasks;
+    private final FileScanTaskReader<RowData> fileReader;
+
+    private LookupIterable(
+        CloseableIterable<CombinedScanTask> tasks, FileScanTaskReader<RowData> 
fileReader) {
+      this.tasks = tasks;
+      this.fileReader = fileReader;
+    }
+
+    @Override
+    public CloseableIterator<RowData> iterator() {
+      CloseableIterator<RowData> rows =
+          CloseableIterable.concat(Iterables.transform(tasks, 
this::taskRows)).iterator();
+      addCloseable(rows);
+      return rows;
+    }
+
+    @Override
+    public void close() throws IOException {
+      tasks.close();
+      super.close();
+    }
+
+    private CloseableIterable<RowData> taskRows(CombinedScanTask task) {
+      DataIterator<RowData> rows =
+          new DataIterator<>(fileReader, task, table.io(), table.encryption());
+      return CloseableIterable.combine(() -> rows, rows);
+    }
+  }
+}
diff --git 
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
new file mode 100644
index 0000000000..b6e1a4ebc2
--- /dev/null
+++ 
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+class InMemoryLookupCache implements Closeable {
+
+  private final Map<BinaryRowData, List<RowData>> cache;
+
+  InMemoryLookupCache() {
+    this.cache = Maps.newHashMap();
+  }
+
+  @Nullable
+  List<RowData> get(BinaryRowData key) {
+    return cache.get(key);
+  }
+
+  void add(BinaryRowData key, BinaryRowData row) {
+    cache.compute(key, (k, existing) -> append(existing, row));
+  }
+
+  private static List<RowData> append(@Nullable List<RowData> existing, 
RowData row) {
+    if (existing == null) {
+      return Collections.singletonList(row);
+    }
+
+    if (existing instanceof ArrayList) {
+      existing.add(row);
+      return existing;
+    }
+
+    List<RowData> rows = Lists.newArrayListWithCapacity(2);
+    rows.add(existing.get(0));
+    rows.add(row);
+    return rows;
+  }
+
+  @Override
+  public void close() {
+    cache.clear();
+  }
+}
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
index 2dc5bc5c65..4a3d585e9d 100644
--- 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
@@ -37,7 +37,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 /** Use the FlinkSource */
-public class TestFlinkSourceSql extends TestSqlBase {
+public class TestFlinkSourceSql extends TestSourceSqlBase {
   @BeforeEach
   @Override
   public void before() throws IOException {
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
new file mode 100644
index 0000000000..40c13af6d4
--- /dev/null
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
@@ -0,0 +1,369 @@
+/*
+ * 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.iceberg.flink.source;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Tests SQL lookup join with Iceberg table source. */
+class TestIcebergLookupJoinSql extends TestSqlBase {
+  private static final Schema DIM_SCHEMA =
+      new Schema(
+          required(1, "user_id", Types.LongType.get()),
+          required(2, "name", Types.StringType.get()),
+          required(3, "city", Types.StringType.get()));
+
+  @BeforeEach
+  @Override
+  public void before() throws IOException {
+    setUpTableEnv(getStreamingTableEnv());
+  }
+
+  @Test
+  void lookupJoinReturnsMatchedAndUnmatchedRows() throws Exception {
+    createDefaultDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders", 5);
+
+    String joinSql = lookupJoinSql("o.order_id, o.user_id, u.name, u.city", 
"orders");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(2L, 2L, "bob", "shanghai"),
+            Row.of(3L, 3L, "carol", "guangzhou"),
+            Row.of(4L, 4L, null, null),
+            Row.of(5L, 5L, null, null));
+  }
+
+  @Test
+  void lookupJoinWithFilterConditionInJoin() throws Exception {
+    Table dimTable =
+        CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
DIM_SCHEMA);
+    GenericAppenderHelper helper =
+        new GenericAppenderHelper(dimTable, FileFormat.PARQUET, 
temporaryFolder);
+    helper.appendToTable(
+        helper.writeFile(
+            Lists.newArrayList(
+                dimRecord(1L, "alice", "beijing"), dimRecord(3L, "carol", 
"beijing"))));
+    DataFile shanghaiFile = helper.writeFile(Lists.newArrayList(dimRecord(2L, 
"bob", "shanghai")));
+    helper.appendToTable(shanghaiFile);
+
+    // The filter must be pushed to the dimension table scan, which skips this 
file by its
+    // column stats. Otherwise loading the cache fails on the missing file.
+    dimTable.io().deleteFile(shanghaiFile.location());
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_with_filter", 4);
+
+    String joinSql =
+        lookupJoinSql(
+            "o.order_id, o.user_id, u.name, u.city",
+            "orders_with_filter",
+            null,
+            "u.city = 'beijing'");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(2L, 2L, null, null),
+            Row.of(3L, 3L, "carol", "beijing"),
+            Row.of(4L, 4L, null, null));
+  }
+
+  @Test
+  void lookupJoinWithDuplicateKeysInDimTable() throws Exception {
+    // The dimension table has two rows sharing the same join key, so every 
probe row
+    // matching that key must be joined with both of them.
+    createDimTable(
+        dimRecord(1L, "alice", "beijing"),
+        dimRecord(1L, "alice-2", "shanghai"),
+        dimRecord(2L, "bob", "shanghai"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_duplicate_keys", 3);
+
+    String joinSql =
+        lookupJoinSql("o.order_id, o.user_id, u.name, u.city", 
"orders_duplicate_keys");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(1L, 1L, "alice-2", "shanghai"),
+            Row.of(2L, 2L, "bob", "shanghai"),
+            Row.of(3L, 3L, null, null));
+  }
+
+  @Test
+  void lookupJoinWithOptionsInTableDdl() throws Exception {
+    createDefaultDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_ddl", 3);
+
+    SqlHelpers.sql(
+        streamEnv,
+        "CREATE TABLE default_catalog.default_database.dim_users_ddl (\n"
+            + "  user_id BIGINT,\n"
+            + "  name    STRING,\n"
+            + "  city    STRING\n"
+            + ") WITH (\n"
+            + "  'connector' = 'iceberg',\n"
+            + "  'catalog-name' = 'iceberg_catalog',\n"
+            + "  'catalog-type' = 'hadoop',\n"
+            + "  'warehouse' = '%s',\n"
+            + "  'catalog-database' = '%s',\n"
+            + "  'catalog-table' = '%s',\n"
+            + "  'lookup.cache' = 'FULL'\n"
+            + ")",
+        CATALOG_EXTENSION.warehouse(),
+        TestFixtures.DATABASE,
+        TestFixtures.TABLE);
+
+    String joinSql =
+        "SELECT o.order_id, o.user_id, u.name, u.city\n"
+            + "FROM orders_ddl AS o\n"
+            + "LEFT JOIN default_catalog.default_database.dim_users_ddl\n"
+            + "  FOR SYSTEM_TIME AS OF o.proc_time AS u\n"
+            + "  ON o.user_id = u.user_id";
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(2L, 2L, "bob", "shanghai"),
+            Row.of(3L, 3L, "carol", "guangzhou"));
+  }
+
+  @Test
+  void lookupJoinAcceptsFullCacheType() throws Exception {
+    createDimTable(dimRecord(1L, "alice", "beijing"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_full_cache_options", 1);
+
+    String joinSql =
+        lookupJoinSql(
+            "o.order_id, u.name",
+            "orders_full_cache_options",
+            "OPTIONS('lookup.cache'='FULL')",
+            null);
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql)).containsExactly(Row.of(1L, 
"alice"));
+  }
+
+  @Test
+  void lookupJoinRejectsUnsupportedCacheType() throws Exception {
+    createDimTable(dimRecord(1L, "alice", "beijing"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_unsupported_cache", 1);
+
+    String joinSql =
+        lookupJoinSql(
+            "o.order_id, u.name",
+            "orders_unsupported_cache",
+            "OPTIONS('lookup.cache'='PARTIAL')",
+            null);
+
+    assertThatThrownBy(() -> SqlHelpers.sql(streamEnv, joinSql))
+        .as("An Iceberg table cannot be point-looked-up, so PARTIAL must be 
rejected")
+        .hasStackTraceContaining("only supports lookup.cache=FULL");
+  }
+
+  @Test
+  void lookupJoinLoadsCacheEagerlyByDefault() throws Exception {
+    String dataFileLocation = createEmptyProbeAndUnreadableDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+
+    assertThatThrownBy(
+            () -> SqlHelpers.sql(streamEnv, lookupJoinSql("o.user_id, u.name", 
"probe_empty")))
+        .as("The cache is loaded when the lookup function is opened, so the 
job fails at startup")
+        .rootCause()
+        .hasMessageContaining(dataFileLocation);
+  }
+
+  @Test
+  void lookupJoinLoadsCacheLazilyWhenEagerLoadDisabled() throws Exception {
+    createEmptyProbeAndUnreadableDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+
+    assertThat(
+            SqlHelpers.sql(
+                streamEnv,
+                lookupJoinSql(
+                    "o.user_id, u.name",
+                    "probe_empty",
+                    "OPTIONS('lookup.full-cache.eager-load'='false')",
+                    null)))
+        .as("The cache is loaded on the first lookup, and no probe row is ever 
looked up")
+        .isEmpty();
+  }
+
+  @Test
+  void lookupJoinWithReorderedDimColumns() throws Exception {
+    createDimTable(dimRecord(1L, "alice", "beijing"), dimRecord(2L, "bob", 
"shanghai"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_reordered", 2);
+
+    String joinSql = lookupJoinSql("o.order_id, u.city, u.name", 
"orders_reordered");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(Row.of(1L, "beijing", "alice"), Row.of(2L, 
"shanghai", "bob"));
+  }
+
+  private static String lookupJoinSql(String selectClause, String probeTable) {
+    return lookupJoinSql(selectClause, probeTable, null, null);
+  }
+
+  private static String lookupJoinSql(
+      String selectClause, String probeTable, String options, String 
extraCondition) {
+    return String.format(
+        "SELECT %s\n"
+            + "FROM %s AS o\n"
+            + "LEFT JOIN iceberg_catalog.`%s`.`%s`\n"
+            + "%s"
+            + "  FOR SYSTEM_TIME AS OF o.proc_time AS u\n"
+            + "  ON o.user_id = u.user_id%s",
+        selectClause,
+        probeTable,
+        TestFixtures.DATABASE,
+        TestFixtures.TABLE,
+        options == null ? "" : "  /*+ " + options + " */\n",
+        extraCondition == null ? "" : " AND " + extraCondition);
+  }
+
+  private void createDefaultDimTable() throws IOException {
+    createDimTable(
+        dimRecord(1L, "alice", "beijing"),
+        dimRecord(2L, "bob", "shanghai"),
+        dimRecord(3L, "carol", "guangzhou"));
+  }
+
+  private void createDimTable(Record... records) throws IOException {
+    Table dimTable =
+        CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
DIM_SCHEMA);
+    GenericAppenderHelper helper =
+        new GenericAppenderHelper(dimTable, FileFormat.PARQUET, 
temporaryFolder);
+    DataFile dataFile = helper.writeFile(Lists.newArrayList(records));
+    helper.appendToTable(dataFile);
+  }
+
+  private static void createOrdersTable(
+      TableEnvironment tableEnvironment, String tableName, int numberOfRows) {
+    SqlHelpers.sql(
+        tableEnvironment,
+        "CREATE TEMPORARY TABLE %s (\n"
+            + "  order_id BIGINT,\n"
+            + "  user_id  BIGINT,\n"
+            + "  proc_time AS PROCTIME()\n"
+            + ") WITH (\n"
+            + "  'connector' = 'datagen',\n"
+            + "  'number-of-rows' = '%d',\n"
+            + "  'fields.order_id.kind' = 'sequence',\n"
+            + "  'fields.order_id.start' = '1',\n"
+            + "  'fields.order_id.end' = '%d',\n"
+            + "  'fields.user_id.kind' = 'sequence',\n"
+            + "  'fields.user_id.start' = '1',\n"
+            + "  'fields.user_id.end' = '%d'\n"
+            + ")",
+        tableName,
+        numberOfRows,
+        numberOfRows,
+        numberOfRows);
+  }
+
+  private static Record dimRecord(long userId, String name, String city) {
+    Record record = GenericRecord.create(DIM_SCHEMA);
+    record.setField("user_id", userId);
+    record.setField("name", name);
+    record.setField("city", city);
+    return record;
+  }
+
+  private static void setUpTableEnv(TableEnvironment tableEnvironment) {
+    Configuration tableConf = tableEnvironment.getConfig().getConfiguration();
+    tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE, 
true);
+    
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, 
false);
+
+    
tableEnvironment.getConfig().set("table.exec.resource.default-parallelism", 
"1");
+    SqlHelpers.sql(
+        tableEnvironment,
+        "create catalog iceberg_catalog with ('type'='iceberg', 
'catalog-type'='hadoop', 'warehouse'='%s')",
+        CATALOG_EXTENSION.warehouse());
+    SqlHelpers.sql(tableEnvironment, "use catalog iceberg_catalog");
+  }
+
+  private String createEmptyProbeAndUnreadableDimTable() throws IOException {
+    Table dimTable =
+        CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
DIM_SCHEMA);
+    GenericAppenderHelper helper =
+        new GenericAppenderHelper(dimTable, FileFormat.PARQUET, 
temporaryFolder);
+    DataFile dataFile = helper.writeFile(Lists.newArrayList(dimRecord(1L, 
"alice", "beijing")));
+    helper.appendToTable(dataFile);
+
+    CATALOG_EXTENSION
+        .catalog()
+        .createTable(TableIdentifier.of(TestFixtures.DATABASE, "probe_empty"), 
DIM_SCHEMA);
+
+    SqlHelpers.sql(
+        getStreamingTableEnv(),
+        "CREATE TEMPORARY TABLE probe_empty (\n"
+            + "  user_id BIGINT,\n"
+            + "  name    STRING,\n"
+            + "  city    STRING,\n"
+            + "  proc_time AS PROCTIME()\n"
+            + ") WITH (\n"
+            + "  'connector' = 'iceberg',\n"
+            + "  'catalog-name' = 'iceberg_catalog',\n"
+            + "  'catalog-type' = 'hadoop',\n"
+            + "  'warehouse' = '%s',\n"
+            + "  'catalog-database' = '%s',\n"
+            + "  'catalog-table' = 'probe_empty'\n"
+            + ")",
+        CATALOG_EXTENSION.warehouse(),
+        TestFixtures.DATABASE);
+
+    dimTable.io().deleteFile(dataFile.location());
+    return dataFile.location();
+  }
+}
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
index 6cfc18868a..7d52662245 100644
--- 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
@@ -46,7 +46,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 /** Use the IcebergSource (FLIP-27) */
-public class TestIcebergSourceSql extends TestSqlBase {
+public class TestIcebergSourceSql extends TestSourceSqlBase {
   private static final Schema SCHEMA_TS =
       new Schema(
           required(1, "t1", Types.TimestampType.withoutZone()),
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
similarity index 74%
copy from 
flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
copy to 
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
index fe4c327310..61ae30bbe1 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
@@ -18,17 +18,11 @@
  */
 package org.apache.iceberg.flink.source;
 
-import static org.apache.iceberg.flink.TestFixtures.DATABASE;
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.io.IOException;
-import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.test.junit5.MiniClusterExtension;
 import org.apache.flink.types.Row;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.FileFormat;
@@ -38,60 +32,14 @@ import org.apache.iceberg.data.GenericAppenderHelper;
 import org.apache.iceberg.data.RandomGenericData;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.flink.FlinkConfigOptions;
-import org.apache.iceberg.flink.HadoopCatalogExtension;
-import org.apache.iceberg.flink.MiniFlinkClusterExtension;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.TestFixtures;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-import org.junit.jupiter.api.io.TempDir;
 
 /** Test other more advanced usage of SQL. They don't need to run for every 
file format. */
-public abstract class TestSqlBase {
-  @RegisterExtension
-  public static MiniClusterExtension miniClusterExtension =
-      MiniFlinkClusterExtension.createWithClassloaderCheckDisabled();
-
-  @RegisterExtension
-  public static final HadoopCatalogExtension CATALOG_EXTENSION =
-      new HadoopCatalogExtension(DATABASE, TestFixtures.TABLE);
-
-  @TempDir protected Path temporaryFolder;
-
-  private volatile TableEnvironment tEnv;
-
-  private volatile TableEnvironment streamingTEnv;
-
-  protected TableEnvironment getTableEnv() {
-    if (tEnv == null) {
-      synchronized (this) {
-        if (tEnv == null) {
-          this.tEnv =
-              
TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
-        }
-      }
-    }
-    return tEnv;
-  }
-
-  protected TableEnvironment getStreamingTableEnv() {
-    if (streamingTEnv == null) {
-      synchronized (this) {
-        if (streamingTEnv == null) {
-          this.streamingTEnv =
-              
TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());
-        }
-      }
-    }
-
-    return streamingTEnv;
-  }
-
-  @BeforeEach
-  public abstract void before() throws IOException;
+public abstract class TestSourceSqlBase extends TestSqlBase {
 
   @Test
   public void testResiduals() throws Exception {
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
index dd63154fe0..1ad17af2f3 100644
--- 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
@@ -19,37 +19,20 @@
 package org.apache.iceberg.flink.source;
 
 import static org.apache.iceberg.flink.TestFixtures.DATABASE;
-import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.IOException;
 import java.nio.file.Path;
-import java.util.List;
-import java.util.Map;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.apache.flink.types.Row;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.TestHelpers;
-import org.apache.iceberg.data.GenericAppenderHelper;
-import org.apache.iceberg.data.RandomGenericData;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.flink.FlinkConfigOptions;
 import org.apache.iceberg.flink.HadoopCatalogExtension;
 import org.apache.iceberg.flink.MiniFlinkClusterExtension;
-import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.TestFixtures;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
 
-/** Test other more advanced usage of SQL. They don't need to run for every 
file format. */
+/** Base class for SQL tests, providing a mini cluster, a Hadoop catalog and 
table environments. */
 public abstract class TestSqlBase {
   @RegisterExtension
   public static MiniClusterExtension miniClusterExtension =
@@ -92,84 +75,4 @@ public abstract class TestSqlBase {
 
   @BeforeEach
   public abstract void before() throws IOException;
-
-  @Test
-  public void testResiduals() throws Exception {
-    Table table =
-        CATALOG_EXTENSION
-            .catalog()
-            .createTable(TestFixtures.TABLE_IDENTIFIER, TestFixtures.SCHEMA, 
TestFixtures.SPEC);
-
-    List<Record> writeRecords = 
RandomGenericData.generate(TestFixtures.SCHEMA, 2, 0L);
-    writeRecords.get(0).set(1, 123L);
-    writeRecords.get(0).set(2, "2020-03-20");
-    writeRecords.get(1).set(1, 456L);
-    writeRecords.get(1).set(2, "2020-03-20");
-
-    GenericAppenderHelper helper =
-        new GenericAppenderHelper(table, FileFormat.PARQUET, temporaryFolder);
-
-    List<Record> expectedRecords = Lists.newArrayList();
-    expectedRecords.add(writeRecords.get(0));
-
-    DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), 
writeRecords);
-    DataFile dataFile2 =
-        helper.writeFile(
-            TestHelpers.Row.of("2020-03-21", 0),
-            RandomGenericData.generate(TestFixtures.SCHEMA, 2, 0L));
-    helper.appendToTable(dataFile1, dataFile2);
-
-    org.apache.iceberg.flink.TestHelpers.assertRecords(
-        run(Maps.newHashMap(), "where dt='2020-03-20' and id=123", "*"),
-        expectedRecords,
-        TestFixtures.SCHEMA);
-  }
-
-  @Test
-  public void testExposeLocality() throws Exception {
-    Table table =
-        CATALOG_EXTENSION
-            .catalog()
-            .createTable(TestFixtures.TABLE_IDENTIFIER, TestFixtures.SCHEMA, 
TestFixtures.SPEC);
-
-    TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
-    List<Record> expectedRecords = 
RandomGenericData.generate(TestFixtures.SCHEMA, 10, 0L);
-    expectedRecords.forEach(expectedRecord -> expectedRecord.set(2, 
"2020-03-20"));
-
-    GenericAppenderHelper helper =
-        new GenericAppenderHelper(table, FileFormat.PARQUET, temporaryFolder);
-    DataFile dataFile = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), 
expectedRecords);
-    helper.appendToTable(dataFile);
-
-    // test sql api
-    Configuration tableConf = getTableEnv().getConfig().getConfiguration();
-    tableConf.setBoolean(
-        
FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO.key(), false);
-
-    List<Row> results = SqlHelpers.sql(getTableEnv(), "select * from t");
-    org.apache.iceberg.flink.TestHelpers.assertRecords(
-        results, expectedRecords, TestFixtures.SCHEMA);
-
-    // test table api
-    tableConf.setBoolean(
-        
FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO.key(), true);
-    FlinkSource.Builder builder = 
FlinkSource.forRowData().tableLoader(tableLoader).table(table);
-
-    // When running with CI or local, `localityEnabled` will be false even if 
this configuration is
-    // enabled
-    assertThat(SourceUtil.isLocalityEnabled(table, tableConf, true))
-        .as("Expose split locality info should be false.")
-        .isFalse();
-
-    results = run(Maps.newHashMap(), "where dt='2020-03-20'", "*");
-    org.apache.iceberg.flink.TestHelpers.assertRecords(
-        results, expectedRecords, TestFixtures.SCHEMA);
-  }
-
-  protected List<Row> run(
-      Map<String, String> options, String sqlFilter, String... 
sqlSelectedFields) {
-    String select = String.join(",", sqlSelectedFields);
-    String optionStr = SqlHelpers.sqlOptionsToString(options);
-    return SqlHelpers.sql(getTableEnv(), "select %s from t %s %s", select, 
optionStr, sqlFilter);
-  }
 }
diff --git 
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
new file mode 100644
index 0000000000..d13d89d17a
--- /dev/null
+++ 
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
@@ -0,0 +1,318 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+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.functions.FunctionContext;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestIcebergLookup {
+  private static final Schema SCHEMA =
+      new Schema(
+          optional(1, "id", Types.LongType.get()),
+          optional(2, "data", Types.StringType.get()),
+          optional(3, "category", Types.StringType.get()));
+
+  private static final String[] PROJECTED_COLUMNS = {"id", "data", "category"};
+  private static final RowType ROW_TYPE = FlinkSchemaUtil.convert(SCHEMA);
+  private static final int[] ID_KEY_INDEXES = {0};
+  private static final int[] ID_AND_CATEGORY_KEY_INDEXES = {0, 2};
+
+  @TempDir private Path temporaryFolder;
+
+  @RegisterExtension
+  protected static final HadoopCatalogExtension CATALOG_EXTENSION =
+      new HadoopCatalogExtension(TestFixtures.DATABASE, TestFixtures.TABLE);
+
+  private IcebergFullCachingLookupFunction lookupFunction;
+
+  @AfterEach
+  void after() throws Exception {
+    if (lookupFunction != null) {
+      lookupFunction.close();
+    }
+  }
+
+  @Test
+  void lookupReaderReadsWithBaseFilters() throws Exception {
+    Table table = createTableWithRecords();
+    IcebergLookupReader reader =
+        lookupReader(table, ImmutableList.of(Expressions.equal("category", 
"B")), false);
+
+    List<List<Object>> rows = Lists.newArrayList();
+    try (CloseableIterable<RowData> rowData = 
reader.read(IcebergLookupReader.CURRENT_SNAPSHOT)) {
+      for (RowData row : rowData) {
+        rows.add(
+            ImmutableList.of(
+                row.getLong(0), row.getString(1).toString(), 
row.getString(2).toString()));
+      }
+    }
+
+    assertThat(rows).containsExactly(ImmutableList.of(2L, "bob", "B"));
+  }
+
+  @Test
+  void lookupReaderRespectsCaseSensitivity() throws Exception {
+    Table table = createTableWithRecords();
+    List<Expression> filters = ImmutableList.of(Expressions.equal("CATEGORY", 
"B"));
+
+    IcebergLookupReader caseSensitiveReader = lookupReader(table, filters, 
true);
+    assertThatThrownBy(() -> readIds(caseSensitiveReader, 
IcebergLookupReader.CURRENT_SNAPSHOT))
+        .as("Case sensitive lookup should reject a filter that doesn't match 
the column case")
+        .isInstanceOf(ValidationException.class)
+        .hasMessageContaining("CATEGORY");
+
+    assertThat(readIds(lookupReader(table, filters, false), 
IcebergLookupReader.CURRENT_SNAPSHOT))
+        .containsExactly(2L);
+  }
+
+  @Test
+  void lookupReaderReadsPinnedSnapshot() throws Exception {
+    Table table = createTableWithRecords();
+    long pinnedSnapshot = table.currentSnapshot().snapshotId();
+
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    IcebergLookupReader reader = lookupReader(table, ImmutableList.of(), 
false);
+
+    assertThat(readIds(reader, pinnedSnapshot))
+        .as("Pinned snapshot should not contain the appended row")
+        .containsExactlyInAnyOrder(1L, 2L, 3L);
+    assertThat(readIds(reader, IcebergLookupReader.CURRENT_SNAPSHOT))
+        .as("Current snapshot should contain the appended row")
+        .containsExactlyInAnyOrder(1L, 2L, 3L, 6L);
+  }
+
+  @Test
+  void lookupFunctionReturnsRowsFromCache() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    assertThat(lookupFunction.lookup(keyRow(6L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(404L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+  }
+
+  @Test
+  void lookupFunctionWithMultiColumnKey() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_AND_CATEGORY_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L, "A")))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+    assertThat(lookupFunction.lookup(keyRow(1L, "B"))).isEmpty();
+
+    appendRecords(table, ImmutableList.of(record(3L, "carol-2", "A")));
+
+    assertThat(lookupFunction.lookup(keyRow(3L, "A")))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 3L, "carol", "A"));
+  }
+
+  @Test
+  void lookupFunctionLoadsCacheLazily() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    // Nothing is loaded yet, so the first lookup sees the row appended after 
opening.
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    assertThat(lookupFunction.lookup(keyRow(6L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 6L, "frank", "D"));
+  }
+
+  @Test
+  void lookupFunctionLoadsCacheEagerly() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, true);
+    lookupFunction.open(new FunctionContext(null));
+
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    assertThat(lookupFunction.lookup(keyRow(6L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+  }
+
+  @Test
+  void lookupFunctionHandlesEmptyTable() throws Exception {
+    CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
SCHEMA);
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, true);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L))).isEmpty();
+  }
+
+  @Test
+  void lookupFunctionHandlesMultipleRowsPerKey() throws Exception {
+    Table table = 
CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA);
+    appendRecords(
+        table,
+        ImmutableList.of(
+            record(1L, "alice", "A"), record(1L, "alice-2", "A"), record(2L, 
"bob", "B")));
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .hasSize(2)
+        .allSatisfy(row -> assertThat(row.getLong(0)).isEqualTo(1L));
+    assertThat(lookupFunction.lookup(keyRow(2L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 2L, "bob", "B"));
+  }
+
+  @Test
+  void lookupFunctionAppliesPushedFilters() throws Exception {
+    createTableWithRecords();
+
+    lookupFunction =
+        new IcebergFullCachingLookupFunction(
+            CATALOG_EXTENSION.tableLoader().clone(),
+            ROW_TYPE,
+            ID_KEY_INDEXES,
+            ImmutableList.of(Expressions.equal("category", "B")),
+            true,
+            false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(2L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 2L, "bob", "B"));
+  }
+
+  private IcebergFullCachingLookupFunction newLookupFunction(
+      int[] lookupKeyIndexes, boolean eagerLoad) {
+    return new IcebergFullCachingLookupFunction(
+        CATALOG_EXTENSION.tableLoader().clone(),
+        ROW_TYPE,
+        lookupKeyIndexes,
+        ImmutableList.of(),
+        false,
+        eagerLoad);
+  }
+
+  private static IcebergLookupReader lookupReader(
+      Table table, List<Expression> filters, boolean caseSensitive) {
+    return new IcebergLookupReader(
+        table, SCHEMA.select(PROJECTED_COLUMNS), filters, caseSensitive, null);
+  }
+
+  private static List<Long> readIds(IcebergLookupReader reader, long 
snapshotId)
+      throws IOException {
+    List<Long> ids = Lists.newArrayList();
+    try (CloseableIterable<RowData> rows = reader.read(snapshotId)) {
+      for (RowData row : rows) {
+        ids.add(row.isNullAt(0) ? null : row.getLong(0));
+      }
+    }
+
+    return ids;
+  }
+
+  private Table createTableWithRecords() throws Exception {
+    Table table = 
CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA);
+    appendRecords(
+        table,
+        ImmutableList.of(
+            record(1L, "alice", "A"), record(2L, "bob", "B"), record(3L, 
"carol", "A")));
+    return table;
+  }
+
+  private void appendRecords(Table table, List<Record> records) throws 
Exception {
+    new GenericAppenderHelper(table, FileFormat.PARQUET, 
temporaryFolder).appendToTable(records);
+  }
+
+  private static GenericRecord record(Long id, String data, String category) {
+    GenericRecord record = GenericRecord.create(SCHEMA);
+    record.setField("id", id);
+    record.setField("data", data);
+    record.setField("category", category);
+    return record;
+  }
+
+  private static RowData keyRow(Long id) {
+    GenericRowData row = new GenericRowData(1);
+    row.setField(0, id);
+    return row;
+  }
+
+  private static RowData keyRow(Long id, String category) {
+    GenericRowData row = new GenericRowData(2);
+    row.setField(0, id);
+    row.setField(1, StringData.fromString(category));
+    return row;
+  }
+
+  private static void assertRow(RowData row, long id, String data, String 
category) {
+    assertThat(row.getLong(0)).isEqualTo(id);
+    assertThat(row.getString(1)).hasToString(data);
+    assertThat(row.getString(2)).hasToString(category);
+  }
+}
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
index 6306ee7a0a..0c7244dc23 100644
--- 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java
@@ -39,6 +39,7 @@ import 
org.apache.flink.table.factories.DynamicTableSourceFactory;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.flink.source.IcebergTableSource;
+import org.apache.iceberg.flink.source.lookup.IcebergLookupOptions;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
@@ -132,6 +133,7 @@ public class FlinkDynamicTableFactory
     options.add(FlinkCreateTableOptions.CATALOG_TABLE);
     options.add(FlinkCreateTableOptions.USE_DYNAMIC_ICEBERG_SINK);
     options.add(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+    options.add(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD);
     return options;
   }
 
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
index 40df25b59f..759b316667 100644
--- 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java
@@ -33,22 +33,30 @@ import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.ProviderContext;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.LookupTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
 import 
org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
 import 
org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import 
org.apache.flink.table.connector.source.abilities.SupportsSourceWatermark;
+import org.apache.flink.table.connector.source.lookup.LookupFunctionProvider;
+import org.apache.flink.table.connector.source.lookup.LookupOptions;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.expressions.ResolvedExpression;
 import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.functions.LookupFunction;
 import org.apache.flink.table.legacy.api.TableSchema;
 import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkConfParser;
 import org.apache.iceberg.flink.FlinkConfigOptions;
 import org.apache.iceberg.flink.FlinkFilters;
 import org.apache.iceberg.flink.FlinkReadOptions;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.source.assigner.SplitAssignerType;
+import org.apache.iceberg.flink.source.lookup.IcebergFullCachingLookupFunction;
+import org.apache.iceberg.flink.source.lookup.IcebergLookupOptions;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -61,7 +69,8 @@ public class IcebergTableSource
         SupportsProjectionPushDown,
         SupportsFilterPushDown,
         SupportsLimitPushDown,
-        SupportsSourceWatermark {
+        SupportsSourceWatermark,
+        LookupTableSource {
 
   private int[] projectedFields;
   private Long limit;
@@ -72,6 +81,7 @@ public class IcebergTableSource
   private final Map<String, String> properties;
   private final boolean isLimitPushDown;
   private final ReadableConfig readableConfig;
+  private final FlinkConfParser flinkConfParser;
 
   private IcebergTableSource(IcebergTableSource toCopy) {
     this.loader = toCopy.loader;
@@ -82,6 +92,7 @@ public class IcebergTableSource
     this.limit = toCopy.limit;
     this.filters = toCopy.filters;
     this.readableConfig = toCopy.readableConfig;
+    this.flinkConfParser = toCopy.flinkConfParser;
   }
 
   public IcebergTableSource(
@@ -109,6 +120,7 @@ public class IcebergTableSource
     this.limit = limit;
     this.filters = filters;
     this.readableConfig = readableConfig;
+    this.flinkConfParser = new FlinkConfParser(properties, readableConfig);
   }
 
   @Override
@@ -237,4 +249,50 @@ public class IcebergTableSource
   public String asSummaryString() {
     return "Iceberg table source";
   }
+
+  @Override
+  public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) 
{
+    int[][] lookupKeys = context.getKeys();
+    int[] lookupKeyIndexes = new int[lookupKeys.length];
+    for (int i = 0; i < lookupKeys.length; i++) {
+      Preconditions.checkArgument(
+          lookupKeys[i].length == 1, "Iceberg lookup source doesn't support 
nested lookup key.");
+      lookupKeyIndexes[i] = lookupKeys[i][0];
+    }
+
+    ResolvedSchema projected = getProjectedSchema();
+    RowType projectedRowType = (RowType) 
projected.toPhysicalRowDataType().getLogicalType();
+
+    LookupOptions.LookupCacheType requestedCacheType =
+        flinkConfParser
+            .enumConfParser(LookupOptions.LookupCacheType.class)
+            .option(LookupOptions.CACHE_TYPE.key())
+            .parseOptional();
+    Preconditions.checkArgument(
+        requestedCacheType == null || requestedCacheType == 
LookupOptions.LookupCacheType.FULL,
+        "Iceberg lookup join only supports %s=FULL, but it is set to '%s'. 
NONE and PARTIAL are "
+            + "not supported, because an Iceberg table cannot be 
point-looked-up.",
+        LookupOptions.CACHE_TYPE.key(),
+        requestedCacheType);
+
+    boolean eagerLoad =
+        flinkConfParser
+            .booleanConf()
+            .option(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD.key())
+            
.defaultValue(IcebergLookupOptions.FULL_CACHE_EAGER_LOAD.defaultValue())
+            .parse();
+
+    boolean caseSensitive =
+        flinkConfParser
+            .booleanConf()
+            .option(FlinkReadOptions.CASE_SENSITIVE)
+            .flinkConfig(FlinkReadOptions.CASE_SENSITIVE_OPTION)
+            
.defaultValue(FlinkReadOptions.CASE_SENSITIVE_OPTION.defaultValue())
+            .parse();
+
+    LookupFunction lookupFn =
+        new IcebergFullCachingLookupFunction(
+            loader, projectedRowType, lookupKeyIndexes, filters, 
caseSensitive, eagerLoad);
+    return LookupFunctionProvider.of(lookupFn);
+  }
 }
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
new file mode 100644
index 0000000000..d7e78913e3
--- /dev/null
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergFullCachingLookupFunction.java
@@ -0,0 +1,223 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.LookupFunction;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkRowData;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.io.CloseableIterable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A lookup function that serves every lookup from an in-memory cache of the 
rows of an Iceberg
+ * table, after projection and pushed-down filters are applied.
+ *
+ * <p>The cache is loaded from the current snapshot either when the function 
is opened or on the
+ * first lookup, depending on {@link 
IcebergLookupOptions#FULL_CACHE_EAGER_LOAD}. It is never
+ * refreshed, so later changes to the table are not visible.
+ */
+@Internal
+public class IcebergFullCachingLookupFunction extends LookupFunction {
+
+  private static final Logger LOG = 
LoggerFactory.getLogger(IcebergFullCachingLookupFunction.class);
+
+  private static final String METRIC_GROUP = "cache";
+  private static final long UNINITIALIZED = -1L;
+
+  private final TableLoader tableLoader;
+  private final RowType projectedRowType;
+  private final int[] lookupKeyIndexes;
+  private final List<Expression> pushedFilters;
+  private final boolean caseSensitive;
+  private final boolean eagerLoad;
+
+  private transient Table table;
+  private transient IcebergLookupReader reader;
+  private transient RowDataSerializer rowSerializer;
+  private transient RowDataSerializer keySerializer;
+  private transient RowData.FieldGetter[] keyGetters;
+  private transient InMemoryLookupCache cache;
+
+  private transient Counter missCounter;
+  private transient volatile long currentSnapshotId;
+  private transient volatile long cachedRecords;
+  private transient volatile long latestLoadTimeMs;
+
+  public IcebergFullCachingLookupFunction(
+      TableLoader tableLoader,
+      RowType projectedRowType,
+      int[] lookupKeyIndexes,
+      List<Expression> pushedFilters,
+      boolean caseSensitive,
+      boolean eagerLoad) {
+    this.tableLoader = tableLoader;
+    this.projectedRowType = projectedRowType;
+    this.lookupKeyIndexes = lookupKeyIndexes;
+    this.pushedFilters = pushedFilters;
+    this.caseSensitive = caseSensitive;
+    this.eagerLoad = eagerLoad;
+  }
+
+  @Override
+  public void open(FunctionContext context) throws Exception {
+    super.open(context);
+    LOG.info(
+        "IcebergFullCachingLookupFunction opening, projected fields={}, 
lookupKeyIndexes={}",
+        projectedRowType.getFieldNames(),
+        Arrays.toString(lookupKeyIndexes));
+
+    MetricGroup metricGroup = context.getMetricGroup().addGroup(METRIC_GROUP);
+    this.missCounter = metricGroup.counter("missCount");
+    this.currentSnapshotId = IcebergLookupReader.CURRENT_SNAPSHOT;
+    this.cachedRecords = 0L;
+    this.latestLoadTimeMs = UNINITIALIZED;
+    metricGroup.gauge("snapshotId", () -> currentSnapshotId);
+    metricGroup.gauge("numCachedRecords", () -> cachedRecords);
+    metricGroup.gauge("latestLoadTime", () -> latestLoadTimeMs);
+
+    tableLoader.open();
+    this.table = tableLoader.loadTable();
+
+    Schema icebergProjection =
+        FlinkSchemaUtil.convert(table.schema(), 
FlinkSchemaUtil.toResolvedSchema(projectedRowType));
+
+    createAccessors();
+
+    String nameMapping = 
table.properties().get(TableProperties.DEFAULT_NAME_MAPPING);
+    this.reader =
+        new IcebergLookupReader(
+            table, icebergProjection, pushedFilters, caseSensitive, 
nameMapping);
+
+    if (eagerLoad) {
+      loadCache();
+    }
+  }
+
+  /**
+   * Returns the cached rows for the key. The returned collection and its rows 
are shared with the
+   * cache, so they must not be modified.
+   */
+  @Override
+  public Collection<RowData> lookup(RowData keyRow) throws IOException {
+    if (cache == null) {
+      loadCache();
+    }
+
+    List<RowData> hit = cache.get(keySerializer.toBinaryRow(keyRow));
+    if (hit == null) {
+      missCounter.inc();
+      return Collections.emptyList();
+    }
+
+    return hit;
+  }
+
+  @Override
+  public void close() throws Exception {
+    try {
+      tableLoader.close();
+    } finally {
+      if (cache != null) {
+        cache.close();
+      }
+
+      super.close();
+    }
+  }
+
+  private void loadCache() throws IOException {
+    table.refresh();
+    Snapshot snapshot = table.currentSnapshot();
+    long snapshotId =
+        snapshot == null ? IcebergLookupReader.CURRENT_SNAPSHOT : 
snapshot.snapshotId();
+
+    LOG.info(
+        "IcebergFullCachingLookupFunction loading started, snapshot={}, 
committedAt={}, projected fields={}, pushedFilters={}",
+        snapshot == null ? "none" : snapshotId,
+        snapshot == null ? "n/a" : snapshot.timestampMillis(),
+        projectedRowType.getFieldNames(),
+        pushedFilters);
+
+    InMemoryLookupCache loaded = new InMemoryLookupCache();
+    long rowCnt = 0L;
+    long start = System.currentTimeMillis();
+    try (CloseableIterable<RowData> rows = reader.read(snapshotId)) {
+      for (RowData row : rows) {
+        loaded.add(extractKey(row), rowSerializer.toBinaryRow(row).copy());
+        rowCnt++;
+      }
+    } finally {
+      this.latestLoadTimeMs = System.currentTimeMillis() - start;
+    }
+
+    this.cache = loaded;
+    this.currentSnapshotId = snapshotId;
+    this.cachedRecords = rowCnt;
+
+    LOG.info(
+        "IcebergFullCachingLookupFunction loading finished, snapshot={}, 
rows={}, cost={} ms",
+        snapshot == null ? "none" : snapshotId,
+        rowCnt,
+        latestLoadTimeMs);
+  }
+
+  private BinaryRowData extractKey(RowData row) {
+    GenericRowData key = new GenericRowData(keyGetters.length);
+    for (int i = 0; i < keyGetters.length; i++) {
+      key.setField(i, keyGetters[i].getFieldOrNull(row));
+    }
+
+    return keySerializer.toBinaryRow(key).copy();
+  }
+
+  private void createAccessors() {
+    LogicalType[] keyTypes = new LogicalType[lookupKeyIndexes.length];
+    this.keyGetters = new RowData.FieldGetter[lookupKeyIndexes.length];
+    for (int i = 0; i < lookupKeyIndexes.length; i++) {
+      int projectedIndex = lookupKeyIndexes[i];
+      keyTypes[i] = projectedRowType.getTypeAt(projectedIndex);
+      this.keyGetters[i] = FlinkRowData.createFieldGetter(keyTypes[i], 
projectedIndex);
+    }
+
+    this.rowSerializer = new RowDataSerializer(projectedRowType);
+    this.keySerializer = new RowDataSerializer(keyTypes);
+  }
+}
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
new file mode 100644
index 0000000000..355fd5387f
--- /dev/null
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupOptions.java
@@ -0,0 +1,41 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/** Lookup join options for the Iceberg table source. */
+@Internal
+public class IcebergLookupOptions {
+
+  public static final ConfigOption<Boolean> FULL_CACHE_EAGER_LOAD =
+      ConfigOptions.key("lookup.full-cache.eager-load")
+          .booleanType()
+          .defaultValue(true)
+          .withDescription(
+              "Whether to load the full lookup cache when the lookup function 
is opened, instead "
+                  + "of on the first lookup. Eager loading narrows the window 
in which the subtasks "
+                  + "of the join can end up on different snapshots of the 
dimension table, and "
+                  + "fails the job at startup if the table cannot be read, at 
the cost of a longer "
+                  + "deployment.");
+
+  private IcebergLookupOptions() {}
+}
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
new file mode 100644
index 0000000000..d669178557
--- /dev/null
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/IcebergLookupReader.java
@@ -0,0 +1,124 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import java.io.IOException;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.flink.source.DataIterator;
+import org.apache.iceberg.flink.source.FileScanTaskReader;
+import org.apache.iceberg.flink.source.RowDataFileScanTaskReader;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.util.SnapshotUtil;
+
+/** Reads all rows of an Iceberg table to build a lookup cache. */
+@Internal
+class IcebergLookupReader {
+
+  static final long CURRENT_SNAPSHOT = -1L;
+
+  private final Table table;
+  private final Schema projectedSchema;
+  private final List<Expression> baseFilters;
+  private final Expression filter;
+  private final boolean caseSensitive;
+  private final String nameMapping;
+
+  IcebergLookupReader(
+      Table table,
+      Schema projectedSchema,
+      List<Expression> baseFilters,
+      boolean caseSensitive,
+      @Nullable String nameMapping) {
+    Preconditions.checkNotNull(baseFilters, "Base filters should not be null");
+
+    this.table = table;
+    this.projectedSchema = projectedSchema;
+    this.baseFilters = baseFilters;
+    this.caseSensitive = caseSensitive;
+    this.nameMapping = nameMapping;
+
+    Expression combinedFilter = Expressions.alwaysTrue();
+    for (Expression baseFilter : this.baseFilters) {
+      combinedFilter = Expressions.and(combinedFilter, baseFilter);
+    }
+
+    this.filter = combinedFilter;
+  }
+
+  CloseableIterable<RowData> read(long snapshotId) {
+    Schema tableSchema =
+        snapshotId == CURRENT_SNAPSHOT ? table.schema() : 
SnapshotUtil.schemaFor(table, snapshotId);
+
+    RowDataFileScanTaskReader fileReader =
+        new RowDataFileScanTaskReader(
+            tableSchema, projectedSchema, nameMapping, caseSensitive, 
baseFilters);
+
+    TableScan scan =
+        
table.newScan().caseSensitive(caseSensitive).project(projectedSchema).filter(filter);
+    if (snapshotId != CURRENT_SNAPSHOT) {
+      scan = scan.useSnapshot(snapshotId);
+    }
+
+    return new LookupIterable(scan.planTasks(), fileReader);
+  }
+
+  private class LookupIterable extends CloseableGroup implements 
CloseableIterable<RowData> {
+    private final CloseableIterable<CombinedScanTask> tasks;
+    private final FileScanTaskReader<RowData> fileReader;
+
+    private LookupIterable(
+        CloseableIterable<CombinedScanTask> tasks, FileScanTaskReader<RowData> 
fileReader) {
+      this.tasks = tasks;
+      this.fileReader = fileReader;
+    }
+
+    @Override
+    public CloseableIterator<RowData> iterator() {
+      CloseableIterator<RowData> rows =
+          CloseableIterable.concat(Iterables.transform(tasks, 
this::taskRows)).iterator();
+      addCloseable(rows);
+      return rows;
+    }
+
+    @Override
+    public void close() throws IOException {
+      tasks.close();
+      super.close();
+    }
+
+    private CloseableIterable<RowData> taskRows(CombinedScanTask task) {
+      DataIterator<RowData> rows =
+          new DataIterator<>(fileReader, task, table.io(), table.encryption());
+      return CloseableIterable.combine(() -> rows, rows);
+    }
+  }
+}
diff --git 
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
new file mode 100644
index 0000000000..b6e1a4ebc2
--- /dev/null
+++ 
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/lookup/InMemoryLookupCache.java
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+class InMemoryLookupCache implements Closeable {
+
+  private final Map<BinaryRowData, List<RowData>> cache;
+
+  InMemoryLookupCache() {
+    this.cache = Maps.newHashMap();
+  }
+
+  @Nullable
+  List<RowData> get(BinaryRowData key) {
+    return cache.get(key);
+  }
+
+  void add(BinaryRowData key, BinaryRowData row) {
+    cache.compute(key, (k, existing) -> append(existing, row));
+  }
+
+  private static List<RowData> append(@Nullable List<RowData> existing, 
RowData row) {
+    if (existing == null) {
+      return Collections.singletonList(row);
+    }
+
+    if (existing instanceof ArrayList) {
+      existing.add(row);
+      return existing;
+    }
+
+    List<RowData> rows = Lists.newArrayListWithCapacity(2);
+    rows.add(existing.get(0));
+    rows.add(row);
+    return rows;
+  }
+
+  @Override
+  public void close() {
+    cache.clear();
+  }
+}
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
index 2dc5bc5c65..4a3d585e9d 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkSourceSql.java
@@ -37,7 +37,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 /** Use the FlinkSource */
-public class TestFlinkSourceSql extends TestSqlBase {
+public class TestFlinkSourceSql extends TestSourceSqlBase {
   @BeforeEach
   @Override
   public void before() throws IOException {
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
new file mode 100644
index 0000000000..40c13af6d4
--- /dev/null
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergLookupJoinSql.java
@@ -0,0 +1,369 @@
+/*
+ * 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.iceberg.flink.source;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/** Tests SQL lookup join with Iceberg table source. */
+class TestIcebergLookupJoinSql extends TestSqlBase {
+  private static final Schema DIM_SCHEMA =
+      new Schema(
+          required(1, "user_id", Types.LongType.get()),
+          required(2, "name", Types.StringType.get()),
+          required(3, "city", Types.StringType.get()));
+
+  @BeforeEach
+  @Override
+  public void before() throws IOException {
+    setUpTableEnv(getStreamingTableEnv());
+  }
+
+  @Test
+  void lookupJoinReturnsMatchedAndUnmatchedRows() throws Exception {
+    createDefaultDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders", 5);
+
+    String joinSql = lookupJoinSql("o.order_id, o.user_id, u.name, u.city", 
"orders");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(2L, 2L, "bob", "shanghai"),
+            Row.of(3L, 3L, "carol", "guangzhou"),
+            Row.of(4L, 4L, null, null),
+            Row.of(5L, 5L, null, null));
+  }
+
+  @Test
+  void lookupJoinWithFilterConditionInJoin() throws Exception {
+    Table dimTable =
+        CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
DIM_SCHEMA);
+    GenericAppenderHelper helper =
+        new GenericAppenderHelper(dimTable, FileFormat.PARQUET, 
temporaryFolder);
+    helper.appendToTable(
+        helper.writeFile(
+            Lists.newArrayList(
+                dimRecord(1L, "alice", "beijing"), dimRecord(3L, "carol", 
"beijing"))));
+    DataFile shanghaiFile = helper.writeFile(Lists.newArrayList(dimRecord(2L, 
"bob", "shanghai")));
+    helper.appendToTable(shanghaiFile);
+
+    // The filter must be pushed to the dimension table scan, which skips this 
file by its
+    // column stats. Otherwise loading the cache fails on the missing file.
+    dimTable.io().deleteFile(shanghaiFile.location());
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_with_filter", 4);
+
+    String joinSql =
+        lookupJoinSql(
+            "o.order_id, o.user_id, u.name, u.city",
+            "orders_with_filter",
+            null,
+            "u.city = 'beijing'");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(2L, 2L, null, null),
+            Row.of(3L, 3L, "carol", "beijing"),
+            Row.of(4L, 4L, null, null));
+  }
+
+  @Test
+  void lookupJoinWithDuplicateKeysInDimTable() throws Exception {
+    // The dimension table has two rows sharing the same join key, so every 
probe row
+    // matching that key must be joined with both of them.
+    createDimTable(
+        dimRecord(1L, "alice", "beijing"),
+        dimRecord(1L, "alice-2", "shanghai"),
+        dimRecord(2L, "bob", "shanghai"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_duplicate_keys", 3);
+
+    String joinSql =
+        lookupJoinSql("o.order_id, o.user_id, u.name, u.city", 
"orders_duplicate_keys");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(1L, 1L, "alice-2", "shanghai"),
+            Row.of(2L, 2L, "bob", "shanghai"),
+            Row.of(3L, 3L, null, null));
+  }
+
+  @Test
+  void lookupJoinWithOptionsInTableDdl() throws Exception {
+    createDefaultDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_ddl", 3);
+
+    SqlHelpers.sql(
+        streamEnv,
+        "CREATE TABLE default_catalog.default_database.dim_users_ddl (\n"
+            + "  user_id BIGINT,\n"
+            + "  name    STRING,\n"
+            + "  city    STRING\n"
+            + ") WITH (\n"
+            + "  'connector' = 'iceberg',\n"
+            + "  'catalog-name' = 'iceberg_catalog',\n"
+            + "  'catalog-type' = 'hadoop',\n"
+            + "  'warehouse' = '%s',\n"
+            + "  'catalog-database' = '%s',\n"
+            + "  'catalog-table' = '%s',\n"
+            + "  'lookup.cache' = 'FULL'\n"
+            + ")",
+        CATALOG_EXTENSION.warehouse(),
+        TestFixtures.DATABASE,
+        TestFixtures.TABLE);
+
+    String joinSql =
+        "SELECT o.order_id, o.user_id, u.name, u.city\n"
+            + "FROM orders_ddl AS o\n"
+            + "LEFT JOIN default_catalog.default_database.dim_users_ddl\n"
+            + "  FOR SYSTEM_TIME AS OF o.proc_time AS u\n"
+            + "  ON o.user_id = u.user_id";
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(
+            Row.of(1L, 1L, "alice", "beijing"),
+            Row.of(2L, 2L, "bob", "shanghai"),
+            Row.of(3L, 3L, "carol", "guangzhou"));
+  }
+
+  @Test
+  void lookupJoinAcceptsFullCacheType() throws Exception {
+    createDimTable(dimRecord(1L, "alice", "beijing"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_full_cache_options", 1);
+
+    String joinSql =
+        lookupJoinSql(
+            "o.order_id, u.name",
+            "orders_full_cache_options",
+            "OPTIONS('lookup.cache'='FULL')",
+            null);
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql)).containsExactly(Row.of(1L, 
"alice"));
+  }
+
+  @Test
+  void lookupJoinRejectsUnsupportedCacheType() throws Exception {
+    createDimTable(dimRecord(1L, "alice", "beijing"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_unsupported_cache", 1);
+
+    String joinSql =
+        lookupJoinSql(
+            "o.order_id, u.name",
+            "orders_unsupported_cache",
+            "OPTIONS('lookup.cache'='PARTIAL')",
+            null);
+
+    assertThatThrownBy(() -> SqlHelpers.sql(streamEnv, joinSql))
+        .as("An Iceberg table cannot be point-looked-up, so PARTIAL must be 
rejected")
+        .hasStackTraceContaining("only supports lookup.cache=FULL");
+  }
+
+  @Test
+  void lookupJoinLoadsCacheEagerlyByDefault() throws Exception {
+    String dataFileLocation = createEmptyProbeAndUnreadableDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+
+    assertThatThrownBy(
+            () -> SqlHelpers.sql(streamEnv, lookupJoinSql("o.user_id, u.name", 
"probe_empty")))
+        .as("The cache is loaded when the lookup function is opened, so the 
job fails at startup")
+        .rootCause()
+        .hasMessageContaining(dataFileLocation);
+  }
+
+  @Test
+  void lookupJoinLoadsCacheLazilyWhenEagerLoadDisabled() throws Exception {
+    createEmptyProbeAndUnreadableDimTable();
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+
+    assertThat(
+            SqlHelpers.sql(
+                streamEnv,
+                lookupJoinSql(
+                    "o.user_id, u.name",
+                    "probe_empty",
+                    "OPTIONS('lookup.full-cache.eager-load'='false')",
+                    null)))
+        .as("The cache is loaded on the first lookup, and no probe row is ever 
looked up")
+        .isEmpty();
+  }
+
+  @Test
+  void lookupJoinWithReorderedDimColumns() throws Exception {
+    createDimTable(dimRecord(1L, "alice", "beijing"), dimRecord(2L, "bob", 
"shanghai"));
+
+    TableEnvironment streamEnv = getStreamingTableEnv();
+    createOrdersTable(streamEnv, "orders_reordered", 2);
+
+    String joinSql = lookupJoinSql("o.order_id, u.city, u.name", 
"orders_reordered");
+
+    assertThat(SqlHelpers.sql(streamEnv, joinSql))
+        .containsExactlyInAnyOrder(Row.of(1L, "beijing", "alice"), Row.of(2L, 
"shanghai", "bob"));
+  }
+
+  private static String lookupJoinSql(String selectClause, String probeTable) {
+    return lookupJoinSql(selectClause, probeTable, null, null);
+  }
+
+  private static String lookupJoinSql(
+      String selectClause, String probeTable, String options, String 
extraCondition) {
+    return String.format(
+        "SELECT %s\n"
+            + "FROM %s AS o\n"
+            + "LEFT JOIN iceberg_catalog.`%s`.`%s`\n"
+            + "%s"
+            + "  FOR SYSTEM_TIME AS OF o.proc_time AS u\n"
+            + "  ON o.user_id = u.user_id%s",
+        selectClause,
+        probeTable,
+        TestFixtures.DATABASE,
+        TestFixtures.TABLE,
+        options == null ? "" : "  /*+ " + options + " */\n",
+        extraCondition == null ? "" : " AND " + extraCondition);
+  }
+
+  private void createDefaultDimTable() throws IOException {
+    createDimTable(
+        dimRecord(1L, "alice", "beijing"),
+        dimRecord(2L, "bob", "shanghai"),
+        dimRecord(3L, "carol", "guangzhou"));
+  }
+
+  private void createDimTable(Record... records) throws IOException {
+    Table dimTable =
+        CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
DIM_SCHEMA);
+    GenericAppenderHelper helper =
+        new GenericAppenderHelper(dimTable, FileFormat.PARQUET, 
temporaryFolder);
+    DataFile dataFile = helper.writeFile(Lists.newArrayList(records));
+    helper.appendToTable(dataFile);
+  }
+
+  private static void createOrdersTable(
+      TableEnvironment tableEnvironment, String tableName, int numberOfRows) {
+    SqlHelpers.sql(
+        tableEnvironment,
+        "CREATE TEMPORARY TABLE %s (\n"
+            + "  order_id BIGINT,\n"
+            + "  user_id  BIGINT,\n"
+            + "  proc_time AS PROCTIME()\n"
+            + ") WITH (\n"
+            + "  'connector' = 'datagen',\n"
+            + "  'number-of-rows' = '%d',\n"
+            + "  'fields.order_id.kind' = 'sequence',\n"
+            + "  'fields.order_id.start' = '1',\n"
+            + "  'fields.order_id.end' = '%d',\n"
+            + "  'fields.user_id.kind' = 'sequence',\n"
+            + "  'fields.user_id.start' = '1',\n"
+            + "  'fields.user_id.end' = '%d'\n"
+            + ")",
+        tableName,
+        numberOfRows,
+        numberOfRows,
+        numberOfRows);
+  }
+
+  private static Record dimRecord(long userId, String name, String city) {
+    Record record = GenericRecord.create(DIM_SCHEMA);
+    record.setField("user_id", userId);
+    record.setField("name", name);
+    record.setField("city", city);
+    return record;
+  }
+
+  private static void setUpTableEnv(TableEnvironment tableEnvironment) {
+    Configuration tableConf = tableEnvironment.getConfig().getConfiguration();
+    tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE, 
true);
+    
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_INFER_SOURCE_PARALLELISM, 
false);
+
+    
tableEnvironment.getConfig().set("table.exec.resource.default-parallelism", 
"1");
+    SqlHelpers.sql(
+        tableEnvironment,
+        "create catalog iceberg_catalog with ('type'='iceberg', 
'catalog-type'='hadoop', 'warehouse'='%s')",
+        CATALOG_EXTENSION.warehouse());
+    SqlHelpers.sql(tableEnvironment, "use catalog iceberg_catalog");
+  }
+
+  private String createEmptyProbeAndUnreadableDimTable() throws IOException {
+    Table dimTable =
+        CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
DIM_SCHEMA);
+    GenericAppenderHelper helper =
+        new GenericAppenderHelper(dimTable, FileFormat.PARQUET, 
temporaryFolder);
+    DataFile dataFile = helper.writeFile(Lists.newArrayList(dimRecord(1L, 
"alice", "beijing")));
+    helper.appendToTable(dataFile);
+
+    CATALOG_EXTENSION
+        .catalog()
+        .createTable(TableIdentifier.of(TestFixtures.DATABASE, "probe_empty"), 
DIM_SCHEMA);
+
+    SqlHelpers.sql(
+        getStreamingTableEnv(),
+        "CREATE TEMPORARY TABLE probe_empty (\n"
+            + "  user_id BIGINT,\n"
+            + "  name    STRING,\n"
+            + "  city    STRING,\n"
+            + "  proc_time AS PROCTIME()\n"
+            + ") WITH (\n"
+            + "  'connector' = 'iceberg',\n"
+            + "  'catalog-name' = 'iceberg_catalog',\n"
+            + "  'catalog-type' = 'hadoop',\n"
+            + "  'warehouse' = '%s',\n"
+            + "  'catalog-database' = '%s',\n"
+            + "  'catalog-table' = 'probe_empty'\n"
+            + ")",
+        CATALOG_EXTENSION.warehouse(),
+        TestFixtures.DATABASE);
+
+    dimTable.io().deleteFile(dataFile.location());
+    return dataFile.location();
+  }
+}
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
index 6cfc18868a..7d52662245 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java
@@ -46,7 +46,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 /** Use the IcebergSource (FLIP-27) */
-public class TestIcebergSourceSql extends TestSqlBase {
+public class TestIcebergSourceSql extends TestSourceSqlBase {
   private static final Schema SCHEMA_TS =
       new Schema(
           required(1, "t1", Types.TimestampType.withoutZone()),
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
similarity index 74%
copy from 
flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
copy to 
flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
index fe4c327310..61ae30bbe1 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSourceSqlBase.java
@@ -18,17 +18,11 @@
  */
 package org.apache.iceberg.flink.source;
 
-import static org.apache.iceberg.flink.TestFixtures.DATABASE;
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.io.IOException;
-import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.test.junit5.MiniClusterExtension;
 import org.apache.flink.types.Row;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.FileFormat;
@@ -38,60 +32,14 @@ import org.apache.iceberg.data.GenericAppenderHelper;
 import org.apache.iceberg.data.RandomGenericData;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.flink.FlinkConfigOptions;
-import org.apache.iceberg.flink.HadoopCatalogExtension;
-import org.apache.iceberg.flink.MiniFlinkClusterExtension;
 import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.TestFixtures;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-import org.junit.jupiter.api.io.TempDir;
 
 /** Test other more advanced usage of SQL. They don't need to run for every 
file format. */
-public abstract class TestSqlBase {
-  @RegisterExtension
-  public static MiniClusterExtension miniClusterExtension =
-      MiniFlinkClusterExtension.createWithClassloaderCheckDisabled();
-
-  @RegisterExtension
-  public static final HadoopCatalogExtension CATALOG_EXTENSION =
-      new HadoopCatalogExtension(DATABASE, TestFixtures.TABLE);
-
-  @TempDir protected Path temporaryFolder;
-
-  private volatile TableEnvironment tEnv;
-
-  private volatile TableEnvironment streamingTEnv;
-
-  protected TableEnvironment getTableEnv() {
-    if (tEnv == null) {
-      synchronized (this) {
-        if (tEnv == null) {
-          this.tEnv =
-              
TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
-        }
-      }
-    }
-    return tEnv;
-  }
-
-  protected TableEnvironment getStreamingTableEnv() {
-    if (streamingTEnv == null) {
-      synchronized (this) {
-        if (streamingTEnv == null) {
-          this.streamingTEnv =
-              
TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());
-        }
-      }
-    }
-
-    return streamingTEnv;
-  }
-
-  @BeforeEach
-  public abstract void before() throws IOException;
+public abstract class TestSourceSqlBase extends TestSqlBase {
 
   @Test
   public void testResiduals() throws Exception {
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
index fe4c327310..1ad17af2f3 100644
--- 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java
@@ -19,37 +19,20 @@
 package org.apache.iceberg.flink.source;
 
 import static org.apache.iceberg.flink.TestFixtures.DATABASE;
-import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.IOException;
 import java.nio.file.Path;
-import java.util.List;
-import java.util.Map;
-import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.test.junit5.MiniClusterExtension;
-import org.apache.flink.types.Row;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.TestHelpers;
-import org.apache.iceberg.data.GenericAppenderHelper;
-import org.apache.iceberg.data.RandomGenericData;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.flink.FlinkConfigOptions;
 import org.apache.iceberg.flink.HadoopCatalogExtension;
 import org.apache.iceberg.flink.MiniFlinkClusterExtension;
-import org.apache.iceberg.flink.TableLoader;
 import org.apache.iceberg.flink.TestFixtures;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
 
-/** Test other more advanced usage of SQL. They don't need to run for every 
file format. */
+/** Base class for SQL tests, providing a mini cluster, a Hadoop catalog and 
table environments. */
 public abstract class TestSqlBase {
   @RegisterExtension
   public static MiniClusterExtension miniClusterExtension =
@@ -92,82 +75,4 @@ public abstract class TestSqlBase {
 
   @BeforeEach
   public abstract void before() throws IOException;
-
-  @Test
-  public void testResiduals() throws Exception {
-    Table table =
-        CATALOG_EXTENSION
-            .catalog()
-            .createTable(TestFixtures.TABLE_IDENTIFIER, TestFixtures.SCHEMA, 
TestFixtures.SPEC);
-
-    List<Record> writeRecords = 
RandomGenericData.generate(TestFixtures.SCHEMA, 2, 0L);
-    writeRecords.get(0).set(1, 123L);
-    writeRecords.get(0).set(2, "2020-03-20");
-    writeRecords.get(1).set(1, 456L);
-    writeRecords.get(1).set(2, "2020-03-20");
-
-    GenericAppenderHelper helper =
-        new GenericAppenderHelper(table, FileFormat.PARQUET, temporaryFolder);
-
-    List<Record> expectedRecords = Lists.newArrayList();
-    expectedRecords.add(writeRecords.get(0));
-
-    DataFile dataFile1 = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), 
writeRecords);
-    DataFile dataFile2 =
-        helper.writeFile(
-            TestHelpers.Row.of("2020-03-21", 0),
-            RandomGenericData.generate(TestFixtures.SCHEMA, 2, 0L));
-    helper.appendToTable(dataFile1, dataFile2);
-
-    org.apache.iceberg.flink.TestHelpers.assertRecords(
-        run(Maps.newHashMap(), "where dt='2020-03-20' and id=123", "*"),
-        expectedRecords,
-        TestFixtures.SCHEMA);
-  }
-
-  @Test
-  public void testExposeLocality() throws Exception {
-    Table table =
-        CATALOG_EXTENSION
-            .catalog()
-            .createTable(TestFixtures.TABLE_IDENTIFIER, TestFixtures.SCHEMA, 
TestFixtures.SPEC);
-
-    TableLoader tableLoader = TableLoader.fromHadoopTable(table.location());
-    List<Record> expectedRecords = 
RandomGenericData.generate(TestFixtures.SCHEMA, 10, 0L);
-    expectedRecords.forEach(expectedRecord -> expectedRecord.set(2, 
"2020-03-20"));
-
-    GenericAppenderHelper helper =
-        new GenericAppenderHelper(table, FileFormat.PARQUET, temporaryFolder);
-    DataFile dataFile = helper.writeFile(TestHelpers.Row.of("2020-03-20", 0), 
expectedRecords);
-    helper.appendToTable(dataFile);
-
-    // test sql api
-    Configuration tableConf = getTableEnv().getConfig().getConfiguration();
-    
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO, 
false);
-
-    List<Row> results = SqlHelpers.sql(getTableEnv(), "select * from t");
-    org.apache.iceberg.flink.TestHelpers.assertRecords(
-        results, expectedRecords, TestFixtures.SCHEMA);
-
-    // test table api
-    
tableConf.set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO, 
true);
-    FlinkSource.Builder builder = 
FlinkSource.forRowData().tableLoader(tableLoader).table(table);
-
-    // When running with CI or local, `localityEnabled` will be false even if 
this configuration is
-    // enabled
-    assertThat(SourceUtil.isLocalityEnabled(table, tableConf, true))
-        .as("Expose split locality info should be false.")
-        .isFalse();
-
-    results = run(Maps.newHashMap(), "where dt='2020-03-20'", "*");
-    org.apache.iceberg.flink.TestHelpers.assertRecords(
-        results, expectedRecords, TestFixtures.SCHEMA);
-  }
-
-  protected List<Row> run(
-      Map<String, String> options, String sqlFilter, String... 
sqlSelectedFields) {
-    String select = String.join(",", sqlSelectedFields);
-    String optionStr = SqlHelpers.sqlOptionsToString(options);
-    return SqlHelpers.sql(getTableEnv(), "select %s from t %s %s", select, 
optionStr, sqlFilter);
-  }
 }
diff --git 
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
new file mode 100644
index 0000000000..d13d89d17a
--- /dev/null
+++ 
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/source/lookup/TestIcebergLookup.java
@@ -0,0 +1,318 @@
+/*
+ * 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.iceberg.flink.source.lookup;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+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.functions.FunctionContext;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopCatalogExtension;
+import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestIcebergLookup {
+  private static final Schema SCHEMA =
+      new Schema(
+          optional(1, "id", Types.LongType.get()),
+          optional(2, "data", Types.StringType.get()),
+          optional(3, "category", Types.StringType.get()));
+
+  private static final String[] PROJECTED_COLUMNS = {"id", "data", "category"};
+  private static final RowType ROW_TYPE = FlinkSchemaUtil.convert(SCHEMA);
+  private static final int[] ID_KEY_INDEXES = {0};
+  private static final int[] ID_AND_CATEGORY_KEY_INDEXES = {0, 2};
+
+  @TempDir private Path temporaryFolder;
+
+  @RegisterExtension
+  protected static final HadoopCatalogExtension CATALOG_EXTENSION =
+      new HadoopCatalogExtension(TestFixtures.DATABASE, TestFixtures.TABLE);
+
+  private IcebergFullCachingLookupFunction lookupFunction;
+
+  @AfterEach
+  void after() throws Exception {
+    if (lookupFunction != null) {
+      lookupFunction.close();
+    }
+  }
+
+  @Test
+  void lookupReaderReadsWithBaseFilters() throws Exception {
+    Table table = createTableWithRecords();
+    IcebergLookupReader reader =
+        lookupReader(table, ImmutableList.of(Expressions.equal("category", 
"B")), false);
+
+    List<List<Object>> rows = Lists.newArrayList();
+    try (CloseableIterable<RowData> rowData = 
reader.read(IcebergLookupReader.CURRENT_SNAPSHOT)) {
+      for (RowData row : rowData) {
+        rows.add(
+            ImmutableList.of(
+                row.getLong(0), row.getString(1).toString(), 
row.getString(2).toString()));
+      }
+    }
+
+    assertThat(rows).containsExactly(ImmutableList.of(2L, "bob", "B"));
+  }
+
+  @Test
+  void lookupReaderRespectsCaseSensitivity() throws Exception {
+    Table table = createTableWithRecords();
+    List<Expression> filters = ImmutableList.of(Expressions.equal("CATEGORY", 
"B"));
+
+    IcebergLookupReader caseSensitiveReader = lookupReader(table, filters, 
true);
+    assertThatThrownBy(() -> readIds(caseSensitiveReader, 
IcebergLookupReader.CURRENT_SNAPSHOT))
+        .as("Case sensitive lookup should reject a filter that doesn't match 
the column case")
+        .isInstanceOf(ValidationException.class)
+        .hasMessageContaining("CATEGORY");
+
+    assertThat(readIds(lookupReader(table, filters, false), 
IcebergLookupReader.CURRENT_SNAPSHOT))
+        .containsExactly(2L);
+  }
+
+  @Test
+  void lookupReaderReadsPinnedSnapshot() throws Exception {
+    Table table = createTableWithRecords();
+    long pinnedSnapshot = table.currentSnapshot().snapshotId();
+
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    IcebergLookupReader reader = lookupReader(table, ImmutableList.of(), 
false);
+
+    assertThat(readIds(reader, pinnedSnapshot))
+        .as("Pinned snapshot should not contain the appended row")
+        .containsExactlyInAnyOrder(1L, 2L, 3L);
+    assertThat(readIds(reader, IcebergLookupReader.CURRENT_SNAPSHOT))
+        .as("Current snapshot should contain the appended row")
+        .containsExactlyInAnyOrder(1L, 2L, 3L, 6L);
+  }
+
+  @Test
+  void lookupFunctionReturnsRowsFromCache() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    assertThat(lookupFunction.lookup(keyRow(6L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(404L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+  }
+
+  @Test
+  void lookupFunctionWithMultiColumnKey() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_AND_CATEGORY_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L, "A")))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+
+    assertThat(lookupFunction.lookup(keyRow(1L, "B"))).isEmpty();
+
+    appendRecords(table, ImmutableList.of(record(3L, "carol-2", "A")));
+
+    assertThat(lookupFunction.lookup(keyRow(3L, "A")))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 3L, "carol", "A"));
+  }
+
+  @Test
+  void lookupFunctionLoadsCacheLazily() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    // Nothing is loaded yet, so the first lookup sees the row appended after 
opening.
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    assertThat(lookupFunction.lookup(keyRow(6L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 6L, "frank", "D"));
+  }
+
+  @Test
+  void lookupFunctionLoadsCacheEagerly() throws Exception {
+    Table table = createTableWithRecords();
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, true);
+    lookupFunction.open(new FunctionContext(null));
+
+    appendRecords(table, ImmutableList.of(record(6L, "frank", "D")));
+
+    assertThat(lookupFunction.lookup(keyRow(6L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 1L, "alice", "A"));
+  }
+
+  @Test
+  void lookupFunctionHandlesEmptyTable() throws Exception {
+    CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, 
SCHEMA);
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, true);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L))).isEmpty();
+  }
+
+  @Test
+  void lookupFunctionHandlesMultipleRowsPerKey() throws Exception {
+    Table table = 
CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA);
+    appendRecords(
+        table,
+        ImmutableList.of(
+            record(1L, "alice", "A"), record(1L, "alice-2", "A"), record(2L, 
"bob", "B")));
+
+    lookupFunction = newLookupFunction(ID_KEY_INDEXES, false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L)))
+        .hasSize(2)
+        .allSatisfy(row -> assertThat(row.getLong(0)).isEqualTo(1L));
+    assertThat(lookupFunction.lookup(keyRow(2L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 2L, "bob", "B"));
+  }
+
+  @Test
+  void lookupFunctionAppliesPushedFilters() throws Exception {
+    createTableWithRecords();
+
+    lookupFunction =
+        new IcebergFullCachingLookupFunction(
+            CATALOG_EXTENSION.tableLoader().clone(),
+            ROW_TYPE,
+            ID_KEY_INDEXES,
+            ImmutableList.of(Expressions.equal("category", "B")),
+            true,
+            false);
+    lookupFunction.open(new FunctionContext(null));
+
+    assertThat(lookupFunction.lookup(keyRow(1L))).isEmpty();
+    assertThat(lookupFunction.lookup(keyRow(2L)))
+        .singleElement()
+        .satisfies(row -> assertRow(row, 2L, "bob", "B"));
+  }
+
+  private IcebergFullCachingLookupFunction newLookupFunction(
+      int[] lookupKeyIndexes, boolean eagerLoad) {
+    return new IcebergFullCachingLookupFunction(
+        CATALOG_EXTENSION.tableLoader().clone(),
+        ROW_TYPE,
+        lookupKeyIndexes,
+        ImmutableList.of(),
+        false,
+        eagerLoad);
+  }
+
+  private static IcebergLookupReader lookupReader(
+      Table table, List<Expression> filters, boolean caseSensitive) {
+    return new IcebergLookupReader(
+        table, SCHEMA.select(PROJECTED_COLUMNS), filters, caseSensitive, null);
+  }
+
+  private static List<Long> readIds(IcebergLookupReader reader, long 
snapshotId)
+      throws IOException {
+    List<Long> ids = Lists.newArrayList();
+    try (CloseableIterable<RowData> rows = reader.read(snapshotId)) {
+      for (RowData row : rows) {
+        ids.add(row.isNullAt(0) ? null : row.getLong(0));
+      }
+    }
+
+    return ids;
+  }
+
+  private Table createTableWithRecords() throws Exception {
+    Table table = 
CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA);
+    appendRecords(
+        table,
+        ImmutableList.of(
+            record(1L, "alice", "A"), record(2L, "bob", "B"), record(3L, 
"carol", "A")));
+    return table;
+  }
+
+  private void appendRecords(Table table, List<Record> records) throws 
Exception {
+    new GenericAppenderHelper(table, FileFormat.PARQUET, 
temporaryFolder).appendToTable(records);
+  }
+
+  private static GenericRecord record(Long id, String data, String category) {
+    GenericRecord record = GenericRecord.create(SCHEMA);
+    record.setField("id", id);
+    record.setField("data", data);
+    record.setField("category", category);
+    return record;
+  }
+
+  private static RowData keyRow(Long id) {
+    GenericRowData row = new GenericRowData(1);
+    row.setField(0, id);
+    return row;
+  }
+
+  private static RowData keyRow(Long id, String category) {
+    GenericRowData row = new GenericRowData(2);
+    row.setField(0, id);
+    row.setField(1, StringData.fromString(category));
+    return row;
+  }
+
+  private static void assertRow(RowData row, long id, String data, String 
category) {
+    assertThat(row.getLong(0)).isEqualTo(id);
+    assertThat(row.getString(1)).hasToString(data);
+    assertThat(row.getString(2)).hasToString(category);
+  }
+}
diff --git a/gradle.properties b/gradle.properties
index 874981cfc3..f225d438a4 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -16,7 +16,7 @@
 jmhOutputPath=build/reports/jmh/human-readable-output.txt
 jmhJsonOutputPath=build/reports/jmh/results.json
 jmhIncludeRegex=.*
-systemProp.defaultFlinkVersions=2.3
+systemProp.defaultFlinkVersions=1.20,2.2,2.3
 systemProp.knownFlinkVersions=1.20,2.2,2.3
 systemProp.defaultSparkVersions=4.2
 systemProp.knownSparkVersions=3.5,4.0,4.1,4.2