[flink] Support lookup join for Chain Table

docs/docs/primary-key-table/chain-table.mdx

@@ -95,14 +95,14 @@ CREATE TABLE default.t (
     `t1` STRING,
     `t2` STRING,
     `t3` STRING,
-    `dt` STRING
-) PARTITIONED BY (`dt`) WITH (
+    `date` STRING
+) PARTITIONED BY (`date`) WITH (
   'chain-table.enabled' = 'true',
-  'primary-key' = 'dt,t1',
+  'primary-key' = 'date,t1',
   'sequence.field' = 't2',
   'bucket-key' = 't1',
   'bucket' = '2',
-  'partition.timestamp-pattern' = '$dt', 
+  'partition.timestamp-pattern' = '$date',
   'partition.timestamp-formatter' = 'yyyyMMdd'
 );
 
@@ -130,6 +130,9 @@ Notice that:
 - Chain table is only supported for primary key table, which means you should define `bucket` and `bucket-key` for the table.
 - Chain table should ensure that the schema of each branch is consistent.
 - Deletion vector is not supported for chain table.
+- The delta branch must use the `DEDUPLICATE` merge engine (default) if you plan
+  to use streaming read or lookup join. Other merge engine types are not supported
+  on the delta branch for these incremental read paths. Batch read is not affected.
 
 After creating a chain table, you can read and write data in the following ways.
 
@@ -263,6 +266,51 @@ INSERT INTO downstream_sink SELECT * FROM default.t;
   scan determines which partitions to read based on the chain-merge logic across snapshot and
   delta branches, and applying a partition filter would interfere with this logic. To read a
   specific partition, use batch mode instead.
+- The delta branch must use the `DEDUPLICATE` merge engine (default). Other merge engine
+  types are not supported.
+
+## Lookup Join
+
+Chain tables support Flink lookup joins. When used as a dimension table, the lookup
+join reflects the latest state of the chain table: the most recent snapshot partition
+per group, combined with delta partitions that come after it. Older snapshot
+partitions are considered outdated and excluded. After the initial load, new delta
+branch writes become visible through periodic incremental refresh.
+
+```sql
+SELECT
+    orders.order_id,
+    dim.product_name
+FROM orders
+LEFT JOIN dim /*+ OPTIONS('continuous.discovery-interval' = '5s') */
+    FOR SYSTEM_TIME AS OF orders.proc_time AS dim
+    ON orders.product_id = dim.product_id;
+```
+
+The incremental refresh only monitors the **delta branch**. Writes to the snapshot
+branch are not detected until the lookup join job is restarted.
+
+### Limitations
+
+- The incremental refresh only monitors the **delta branch**. Writes to the snapshot
+  branch are not detected until the lookup join job is restarted.
+- Join key must not contain partition keys. Chain table's partition model spans two
+  branches (snapshot + delta) with anchor-based merging, making partition-level routing
+  unreliable for lookup joins. If partition keys appear in the join condition, an error
+  will be raised.
+- Partition filters are not supported in chain table lookup joins. Specifying a
+  partition filter via the `scan.partitions` table option throws an
+  `UnsupportedOperationException`. This is because the chain table lookup scan
+  determines which partitions to read based on the chain-merge logic across snapshot
+  and delta branches, and applying a partition filter would interfere with this logic.
+- The chain-table-aware lookup scan only supports the default startup mode
+  (`latest-full`). When the user specifies an explicit starting position — such as
+  `scan.snapshot-id`, `scan.timestamp-millis`, `scan.mode = 'latest'`, or
+  `consumer-id` — an `UnsupportedOperationException` is thrown.
+- `lookup.cache` must be `AUTO` or `FULL`. Other cache modes are not supported for
+  chain tables.
+- The delta branch must use the `DEDUPLICATE` merge engine (default). Other merge
+  engine types are not supported.
 
 ## Group Partition
 

paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java

@@ -88,7 +88,7 @@ public ChainKeyValueFileReaderFactory(
     protected TableSchema getDataSchema(DataFileMeta fileMeta) {
         String branch = chainReadContext.fileBranchMapping().get(fileMeta.fileName());
         if (currentBranch.equalsIgnoreCase(branch)) {
-            super.getDataSchema(fileMeta);
+            return super.getDataSchema(fileMeta);
         }
         if (!branchSchemaManagers.containsKey(branch)) {
             throw new RuntimeException("No schema manager found for branch: " + branch);

paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java

@@ -47,6 +47,7 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -66,6 +67,29 @@
  */
 public class ChainGroupReadTable extends FallbackReadFileStoreTable {
 
+    /**
+     * Options that cannot be safely passed to branch tables during copy because they would
+     * interfere with branch-specific configurations. If dynamic options contain any of these, an
+     * error is thrown since we cannot fulfill the copy contract (which expects all options to be
+     * passed through).
+     */
+    private static final Set<String> UNSAFE_COPY_OPTIONS =
+            Collections.unmodifiableSet(
+                    new HashSet<>(
+                            Arrays.asList(
+                                    CoreOptions.MERGE_ENGINE.key(),
+                                    CoreOptions.BUCKET.key(),
+                                    CoreOptions.BUCKET_KEY.key(),
+                                    CoreOptions.SEQUENCE_FIELD.key(),
+                                    CoreOptions.SCAN_MODE.key(),
+                                    CoreOptions.SCAN_SNAPSHOT_ID.key(),
+                                    CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+                                    CoreOptions.SCAN_TIMESTAMP.key(),
+                                    CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+                                    CoreOptions.SCAN_TAG_NAME.key(),
+                                    CoreOptions.SCAN_VERSION.key(),
+                                    CoreOptions.SCAN_WATERMARK.key())));
+
     public ChainGroupReadTable(FileStoreTable snapshotStoreTable, FileStoreTable deltaStoreTable) {
         super(snapshotStoreTable, deltaStoreTable, true);
         checkArgument(snapshotStoreTable instanceof PrimaryKeyFileStoreTable);
@@ -88,22 +112,49 @@ private DataTableScan newDeltaScan() {
 
     @Override
     public FileStoreTable copy(Map<String, String> dynamicOptions) {
-        return new ChainGroupReadTable(
-                wrapped.copy(dynamicOptions), other().copy(rewriteOtherOptions(dynamicOptions)));
+        Map<String, String> wrappedOptions =
+                prepareBranchOptions(dynamicOptions, wrapped.coreOptions().branch(), wrapped);
+
+        Map<String, String> otherOptions =
+                prepareBranchOptions(
+                        rewriteOtherOptions(dynamicOptions),
+                        other().coreOptions().branch(),
+                        other());
+
+        return new ChainGroupReadTable(wrapped.copy(wrappedOptions), other().copy(otherOptions));
     }
 
     @Override
     public FileStoreTable copy(TableSchema newTableSchema) {
+        Map<String, String> wrappedOptions =
+                prepareBranchOptions(
+                        newTableSchema.options(), wrapped.coreOptions().branch(), wrapped);
+
+        Map<String, String> otherOptions =
+                prepareBranchOptions(
+                        rewriteOtherOptions(newTableSchema.options()),
+                        other().coreOptions().branch(),
+                        other());
+
         return new ChainGroupReadTable(
-                wrapped.copy(newTableSchema),
-                other().copy(newTableSchema.copy(rewriteOtherOptions(newTableSchema.options()))));
+                wrapped.copy(newTableSchema.copy(wrappedOptions)),
+                other().copy(newTableSchema.copy(otherOptions)));
     }
 
     @Override
     public FileStoreTable copyWithoutTimeTravel(Map<String, String> dynamicOptions) {
+        Map<String, String> wrappedOptions =
+                prepareBranchOptions(dynamicOptions, wrapped.coreOptions().branch(), wrapped);
+
+        Map<String, String> otherOptions =
+                prepareBranchOptions(
+                        rewriteOtherOptions(dynamicOptions),
+                        other().coreOptions().branch(),
+                        other());
+
         return new ChainGroupReadTable(
-                wrapped.copyWithoutTimeTravel(dynamicOptions),
-                other().copyWithoutTimeTravel(rewriteOtherOptions(dynamicOptions)));
+                wrapped.copyWithoutTimeTravel(wrappedOptions),
+                other().copyWithoutTimeTravel(otherOptions));
     }
 
     @Override
@@ -112,6 +163,30 @@ public FileStoreTable copyWithLatestSchema() {
                 wrapped.copyWithLatestSchema(), other().copyWithLatestSchema());
     }
 
+    /**
+     * Prepares options for a branch table. Starts from the new schema's options, but overrides
+     * branch-owned options (like merge-engine, bucket) with the branch's own values to preserve
+     * branch-specific configurations.
+     */
+    private static Map<String, String> prepareBranchOptions(
+            Map<String, String> newOptions, String branch, FileStoreTable sourceTable) {
+        Map<String, String> result = new HashMap<>(newOptions);
+
+        // Override branch-owned options with sourceTable's values to preserve
+        // branch-specific configurations
+        for (String key : UNSAFE_COPY_OPTIONS) {
+            String sourceValue = sourceTable.schema().options().get(key);
+            if (sourceValue != null) {
+                result.put(key, sourceValue);
+            }
+        }
+
+        // Set branch name (each sub-table has its own branch identity)
+        result.put(CoreOptions.BRANCH.key(), branch);
+
+        return result;
+    }
+
     @Override
     public FileStoreTable switchToBranch(String branchName) {
         return new ChainGroupReadTable(switchWrappedToBranch(branchName), other());

paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java

@@ -40,6 +40,7 @@
 import org.apache.paimon.table.source.snapshot.StartingContext;
 import org.apache.paimon.utils.ChainPartitionProjector;
 import org.apache.paimon.utils.ChainTableUtils;
+import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.SnapshotManager;
 
 import org.slf4j.Logger;
@@ -103,13 +104,17 @@ public class ChainTableStreamScan implements StreamDataTableScan {
     /** Whether the starting plan (Phase 1) has been completed. */
     private boolean startingDone = false;
 
-    /** Predicates and shard for applying to local scans created in {@link #planStarting()}. */
+    /**
+     * Predicates, shard, and bucket filter for applying to local scans in {@link #planStarting()}.
+     */
     private final List<Predicate> predicates = new ArrayList<>();
 
     private int shardIndex = -1;
 
     private int shardCount = -1;
 
+    @Nullable private Filter<Integer> bucketFilter;
+
     /** Maximum number of retries when race condition is detected during position capture. */
     private static final int MAX_RACE_RETRIES = 3;
 
@@ -120,6 +125,8 @@ public ChainTableStreamScan(ChainGroupReadTable chainGroupReadTable) {
                         chainGroupReadTable.schema(), chainGroupReadTable);
         this.deltaStreamScan = (DataTableStreamScan) chainGroupReadTable.other().newStreamScan();
 
+        ChainTableUtils.validateChainTableForIncrementalRead(chainGroupReadTable);
+
         // Initialize partition projector and chain comparator using the established pattern
         // from ChainTableBatchScan.
         List<String> chainKeys =
@@ -223,13 +230,9 @@ private TableScan.Plan planStarting() {
         // 1. Read delta branch data at the pinned snapshot, grouped by partition.
         Map<BinaryRow, List<DataSplit>> deltaSplitsByPartition;
         if (deltaLatestId != null) {
-            FileStoreTable pinnedDelta =
-                    deltaTable.copy(
-                            Collections.singletonMap(
-                                    CoreOptions.SCAN_SNAPSHOT_ID.key(),
-                                    String.valueOf(deltaLatestId)));
+            FileStoreTable pinnedDelta = deltaTable.copy(pinnedOptions(deltaLatestId));
             DataTableScan pinnedDeltaScan = pinnedDelta.newScan();
-            applyPredicatesAndShard(pinnedDeltaScan);
+            applyPredicatesShardAndBucket(pinnedDeltaScan);
             deltaSplitsByPartition = groupByPartition(pinnedDeltaScan);
         } else {
             deltaSplitsByPartition = Collections.emptyMap();
@@ -242,11 +245,7 @@ private TableScan.Plan planStarting() {
         Map<Object, BinaryRow> latestChainPartitionPerGroup = new HashMap<>();
         FileStoreTable pinnedSnapshot = null;
         if (snapshotLatestId != null) {
-            pinnedSnapshot =
-                    chainGroupReadTable.wrapped.copy(
-                            Collections.singletonMap(
-                                    CoreOptions.SCAN_SNAPSHOT_ID.key(),
-                                    String.valueOf(snapshotLatestId)));
+            pinnedSnapshot = chainGroupReadTable.wrapped.copy(pinnedOptions(snapshotLatestId));
             DataTableScan partitionListingScan = pinnedSnapshot.newScan();
             for (BinaryRow partition : partitionListingScan.listPartitions()) {
                 Object groupKey = toGroupKey(partition);
@@ -268,7 +267,7 @@ private TableScan.Plan planStarting() {
         if (!latestPartitions.isEmpty() && pinnedSnapshot != null) {
             DataTableScan snapshotScan = pinnedSnapshot.newScan();
             snapshotScan.withPartitionFilter(latestPartitions);
-            applyPredicatesAndShard(snapshotScan);
+            applyPredicatesShardAndBucket(snapshotScan);
             snapshotSplitsByPartition = groupByPartition(snapshotScan);
         } else {
             snapshotSplitsByPartition = Collections.emptyMap();
@@ -427,17 +426,40 @@ public DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubta
         return this;
     }
 
+    @Override
+    public InnerTableScan withBucketFilter(Filter<Integer> bucketFilter) {
+        this.bucketFilter = bucketFilter;
+        batchScan.withBucketFilter(bucketFilter);
+        deltaStreamScan.withBucketFilter(bucketFilter);
+        return this;
+    }
+
+    /**
+     * Creates options for pinning a branch table to a specific snapshot. Sets {@code
+     * scan.snapshot-id} to the given id and {@code scan.mode=from-snapshot} to ensure the table
+     * reads from the pinned snapshot rather than using its default scan mode.
+     */
+    private static Map<String, String> pinnedOptions(long snapshotId) {
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(snapshotId));
+        options.put(CoreOptions.SCAN_MODE.key(), CoreOptions.StartupMode.FROM_SNAPSHOT.toString());
+        return options;
+    }
+
     /**
-     * Applies all previously set predicates and shard to a newly created scan. Used for the pinned
-     * delta scan in {@link #planStarting()}.
+     * Applies all previously set predicates, shard, and bucket filter to a newly created scan. Used
+     * for the pinned delta scan in {@link #planStarting()}.
      */
-    private void applyPredicatesAndShard(DataTableScan scan) {
+    private void applyPredicatesShardAndBucket(DataTableScan scan) {
         for (Predicate p : predicates) {
             scan.withFilter(p);
         }
         if (shardIndex >= 0) {
             scan.withShard(shardIndex, shardCount);
         }
+        if (bucketFilter != null) {
+            scan.withBucketFilter(bucketFilter);
+        }
     }
 
     @Nullable

paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java

@@ -129,6 +129,22 @@ public int hashCode() {
         return Objects.hash(logicalPartition, dataFiles);
     }
 
+    @Override
+    public String toString() {
+        return "{"
+                + "partition=hash-"
+                + logicalPartition.hashCode()
+                + ", files="
+                + (dataFiles == null ? 0 : dataFiles.size())
+                + ", branches="
+                + fileBranchMapping.size()
+                + ", bucketPaths="
+                + fileBucketPathMapping.size()
+                + '}'
+                + "@"
+                + Integer.toHexString(hashCode());
+    }
+
     private void writeObject(ObjectOutputStream out) throws IOException {
         serialize(new DataOutputViewStreamWrapper(out));
     }

paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java

@@ -364,4 +364,21 @@ public static Predicate createGroupChainPredicate(
 
         return PredicateBuilder.and(conditions);
     }
+
+    /**
+     * Validates that the chain table configuration is compatible with incremental read paths
+     * (streaming read and lookup join). All validation rules for incremental reads should be
+     * centralized here.
+     */
+    public static void validateChainTableForIncrementalRead(ChainGroupReadTable table) {
+        CoreOptions.MergeEngine mergeEngine = table.other().coreOptions().mergeEngine();
+        if (mergeEngine != CoreOptions.MergeEngine.DEDUPLICATE) {
+            throw new IllegalArgumentException(
+                    "Chain table does not support "
+                            + mergeEngine
+                            + " merge engine on the delta branch "
+                            + "for streaming read or lookup join. "
+                            + "Please use the DEDUPLICATE merge engine on the delta branch.");
+        }
+    }
 }