This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 984f359068 [flink] Support lookup join for Chain Table (#8423)
984f359068 is described below
commit 984f359068b4ab9ef7ab7b991b4b79eca08cf111
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Mon Jul 6 20:48:13 2026 +0800
[flink] Support lookup join for Chain Table (#8423)
This PR adds support for Flink lookup join on Chain Table. 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.
---
docs/docs/primary-key-table/chain-table.mdx | 56 +-
.../paimon/io/ChainKeyValueFileReaderFactory.java | 2 +-
.../apache/paimon/table/ChainGroupReadTable.java | 87 +-
.../apache/paimon/table/ChainTableStreamScan.java | 54 +-
.../org/apache/paimon/table/source/ChainSplit.java | 16 +
.../org/apache/paimon/utils/ChainTableUtils.java | 17 +
.../flink/lookup/FileStoreLookupFunction.java | 25 +-
.../paimon/flink/lookup/FullCacheLookupTable.java | 29 +-
.../paimon/flink/lookup/LookupFileStoreTable.java | 53 +
.../paimon/flink/lookup/LookupStreamingReader.java | 38 +-
.../flink/lookup/PrimaryKeyPartialLookupTable.java | 2 +-
.../apache/paimon/flink/FlinkChainTableITCase.java | 1423 +++++++++++++++++++-
12 files changed, 1756 insertions(+), 46 deletions(-)
diff --git a/docs/docs/primary-key-table/chain-table.mdx
b/docs/docs/primary-key-table/chain-table.mdx
index ada14470ec..9556434692 100644
--- a/docs/docs/primary-key-table/chain-table.mdx
+++ b/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
diff --git
a/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
b/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
index d4e98af85e..a06028b2e0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
+++
b/paimon-core/src/main/java/org/apache/paimon/io/ChainKeyValueFileReaderFactory.java
@@ -88,7 +88,7 @@ public class ChainKeyValueFileReaderFactory extends
KeyValueFileReaderFactory {
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);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
index c65c287aec..1428d26e9d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
@@ -47,6 +47,7 @@ import org.apache.paimon.utils.RowDataToObjectArrayConverter;
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 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
*/
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 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
@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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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());
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java
index 276114712a..fcd07cb26b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java
@@ -40,6 +40,7 @@ import org.apache.paimon.table.source.TableScan;
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 class ChainTableStreamScan implements
StreamDataTableScan {
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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
// 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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
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 class ChainTableStreamScan implements
StreamDataTableScan {
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
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
index 6b3512c615..d296b099da 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/ChainSplit.java
@@ -129,6 +129,22 @@ public class ChainSplit implements Split {
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));
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
index e6d4e4aadf..d338df5191 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
@@ -364,4 +364,21 @@ public class ChainTableUtils {
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.");
+ }
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index 69d834615f..6d49c6df6c 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -30,6 +30,8 @@ import org.apache.paimon.flink.utils.RuntimeContextUtils;
import org.apache.paimon.flink.utils.TableScanUtils;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.ChainGroupReadTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.OutOfRangeException;
import org.apache.paimon.types.RowType;
@@ -196,7 +198,12 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
joinKeys);
LOG.info("Creating lookup table for {}.", table.name());
- if (options.get(LOOKUP_CACHE_MODE) == LookupCacheMode.AUTO
+ boolean isChainTable =
+ table instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) table).other()
+ instanceof ChainGroupReadTable;
+ if (!isChainTable
+ && options.get(LOOKUP_CACHE_MODE) == LookupCacheMode.AUTO
&& new HashSet<>(table.primaryKeys()).equals(new
HashSet<>(joinKeys))) {
if (isRemoteServiceAvailable(table)) {
this.lookupTable =
@@ -403,7 +410,21 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
return false;
}
- Long latestSnapshotId = ((FileStoreTable)
table).snapshotManager().latestSnapshotId();
+ boolean isChainTable =
+ table instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) table).other()
+ instanceof ChainGroupReadTable;
+ // For chain tables, use the delta branch's snapshot manager:
+ // table.other() = ChainGroupReadTable, .other() = delta branch
table.
+ Long latestSnapshotId =
+ isChainTable
+ ? ((FallbackReadFileStoreTable)
+ ((FallbackReadFileStoreTable)
table).other())
+ .other()
+ .snapshotManager()
+ .latestSnapshotId()
+ : table.snapshotManager().latestSnapshotId();
+
Long nextSnapshotId = lookupTable.nextSnapshotId();
if (latestSnapshotId == null || nextSnapshotId == null) {
return false;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
index 520021fa4d..66766f518e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -34,6 +34,8 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.sort.BinaryExternalSortBuffer;
+import org.apache.paimon.table.ChainGroupReadTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.RowType;
@@ -258,7 +260,27 @@ public abstract class FullCacheLookupTable implements
LookupTable {
return;
}
- Long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+ // For chain tables, the delta branch maintains its own snapshot
sequence.
+ // Navigate to the delta branch's snapshot manager for backlog
calculation:
+ // unwrapped = ChainTableFileStoreTable
+ // .other() = ChainGroupReadTable
+ // .other() = delta branch table
+ FileStoreTable unwrapped = table;
+ if (table instanceof LookupFileStoreTable) {
+ unwrapped = ((LookupFileStoreTable) table).wrapped();
+ }
+ boolean isChainTable =
+ unwrapped instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) unwrapped).other()
+ instanceof ChainGroupReadTable;
+ Long latestSnapshotId =
+ isChainTable
+ ? ((FallbackReadFileStoreTable)
+ ((FallbackReadFileStoreTable)
unwrapped).other())
+ .other()
+ .snapshotManager()
+ .latestSnapshotId()
+ : table.snapshotManager().latestSnapshotId();
Long nextSnapshotId = reader.nextSnapshotId();
if (latestSnapshotId != null
&& nextSnapshotId != null
@@ -413,7 +435,8 @@ public abstract class FullCacheLookupTable implements
LookupTable {
void finish() throws IOException;
}
- static FullCacheLookupTable create(Context context, long lruCacheSize) {
+ @VisibleForTesting
+ public static FullCacheLookupTable create(Context context, long
lruCacheSize) {
List<String> primaryKeys = context.table.primaryKeys();
if (primaryKeys.isEmpty()) {
return new NoPrimaryKeyLookupTable(context, lruCacheSize);
@@ -445,7 +468,7 @@ public abstract class FullCacheLookupTable implements
LookupTable {
File tempPath,
List<String> joinKey,
@Nullable Set<Integer> requiredCachedBucketIds) {
- this.table = new LookupFileStoreTable(table, joinKey);
+ this.table = LookupFileStoreTable.create(table, joinKey);
this.projection = projection;
this.tablePredicate = tablePredicate;
this.projectedPredicate = projectedPredicate;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupFileStoreTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupFileStoreTable.java
index 353c99d2b1..bde9aa6c02 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupFileStoreTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupFileStoreTable.java
@@ -26,10 +26,13 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.options.description.DescribedEnum;
import org.apache.paimon.options.description.InlineElement;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.ChainGroupReadTable;
import org.apache.paimon.table.DelegatedFileStoreTable;
+import org.apache.paimon.table.FallbackReadFileStoreTable;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.utils.ChainTableUtils;
import java.util.HashSet;
import java.util.List;
@@ -46,6 +49,20 @@ public class LookupFileStoreTable extends
DelegatedFileStoreTable {
private final LookupStreamScanMode lookupScanMode;
+ /**
+ * Creates a {@link LookupFileStoreTable} for the given table. For chain
tables, validates
+ * constraints before returning.
+ */
+ public static LookupFileStoreTable create(FileStoreTable table,
List<String> joinKeys) {
+ if (table instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) table).other() instanceof
ChainGroupReadTable) {
+ ChainGroupReadTable chainGroupReadTable =
+ (ChainGroupReadTable) ((FallbackReadFileStoreTable)
table).other();
+ validateChainTableConstraints(table, chainGroupReadTable,
joinKeys);
+ }
+ return new LookupFileStoreTable(table, joinKeys);
+ }
+
public LookupFileStoreTable(FileStoreTable wrapped, List<String> joinKeys)
{
super(wrapped);
this.lookupScanMode = lookupStreamScanMode(wrapped, joinKeys);
@@ -58,6 +75,10 @@ public class LookupFileStoreTable extends
DelegatedFileStoreTable {
@Override
public InnerTableRead newRead() {
+ if (wrapped instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) wrapped).other() instanceof
ChainGroupReadTable) {
+ return wrapped.newRead();
+ }
switch (lookupScanMode) {
case CHANGELOG:
case FILE_MONITOR:
@@ -73,6 +94,10 @@ public class LookupFileStoreTable extends
DelegatedFileStoreTable {
@Override
public StreamDataTableScan newStreamScan() {
+ if (wrapped instanceof FallbackReadFileStoreTable
+ && ((FallbackReadFileStoreTable) wrapped).other() instanceof
ChainGroupReadTable) {
+ return wrapped.newStreamScan();
+ }
return new LookupDataTableScan(wrapped, wrapped.newSnapshotReader(),
lookupScanMode);
}
@@ -116,6 +141,34 @@ public class LookupFileStoreTable extends
DelegatedFileStoreTable {
}
}
+ private static void validateChainTableConstraints(
+ FileStoreTable table, ChainGroupReadTable chainGroupReadTable,
List<String> joinKeys) {
+ // Validate merge engine early — before scan creation — for better
error messages.
+ // The same validation is also in ChainTableStreamScan constructor as
a safety net.
+
ChainTableUtils.validateChainTableForIncrementalRead(chainGroupReadTable);
+
+ List<String> partitionKeys = table.schema().partitionKeys();
+ for (String joinKey : joinKeys) {
+ if (partitionKeys.contains(joinKey)) {
+ throw new UnsupportedOperationException(
+ "Chain table lookup join does not support partition
keys in join condition. "
+ + "Partition keys: "
+ + partitionKeys
+ + ", join keys: "
+ + joinKeys);
+ }
+ }
+ Options options = Options.fromMap(table.options());
+ FlinkConnectorOptions.LookupCacheMode cacheMode =
options.get(LOOKUP_CACHE_MODE);
+ if (cacheMode != FlinkConnectorOptions.LookupCacheMode.AUTO
+ && cacheMode != FlinkConnectorOptions.LookupCacheMode.FULL) {
+ throw new UnsupportedOperationException(
+ "Chain table lookup join does not support cache mode "
+ + cacheMode
+ + ". Please use AUTO or FULL.");
+ }
+ }
+
/** Inner stream scan mode for lookup table. */
public enum LookupStreamScanMode implements DescribedEnum {
CHANGELOG("changelog", "Streaming reading based on changelog or delta
data files."),
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupStreamingReader.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupStreamingReader.java
index df326054b1..1c2f98fd82 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupStreamingReader.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupStreamingReader.java
@@ -27,10 +27,13 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.ReaderSupplier;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.ChainTableStreamScan;
+import org.apache.paimon.table.source.ChainSplit;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.IncrementalSplit;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamDataTableScan;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.FunctionWithIOException;
@@ -51,6 +54,7 @@ import java.util.stream.IntStream;
import static
org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_BOOTSTRAP_PARALLELISM;
import static
org.apache.paimon.predicate.PredicateBuilder.transformFieldMapping;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** A streaming reader to load data into {@link LookupTable}. */
public class LookupStreamingReader {
@@ -62,7 +66,7 @@ public class LookupStreamingReader {
@Nullable private final Filter<InternalRow> cacheRowFilter;
private final ReadBuilder readBuilder;
@Nullable private final Predicate projectedPredicate;
- private final LookupDataTableScan scan;
+ private final StreamDataTableScan scan;
public LookupStreamingReader(
LookupFileStoreTable table,
@@ -83,8 +87,17 @@ public class LookupStreamingReader {
requireCachedBucketIds == null
? null
: requireCachedBucketIds::contains);
- scan = (LookupDataTableScan) readBuilder.newStreamScan();
- scan.setScanPartitions(scanPartitions);
+ scan = (StreamDataTableScan) readBuilder.newStreamScan();
+ checkArgument(
+ scan instanceof LookupDataTableScan || scan instanceof
ChainTableStreamScan,
+ "Unexpected scan type for lookup: %s",
+ scan.getClass().getName());
+ if (scan instanceof LookupDataTableScan) {
+ ((LookupDataTableScan) scan).setScanPartitions(scanPartitions);
+ }
+ // Note: ChainTableStreamScan does not support partition filters via
setScanPartitions.
+ // Chain tables handle partitions through their own chain-merge logic
across snapshot
+ // and delta branches, so partition-level filtering is not applicable
in the same way.
if (predicate != null) {
List<String> fieldNames = table.rowType().getFieldNames();
@@ -157,14 +170,21 @@ public class LookupStreamingReader {
}
Split split = splits.get(0);
- long snapshotId =
- split instanceof DataSplit
- ? ((DataSplit) split).snapshotId()
- : ((IncrementalSplit) split).snapshotId();
+ String splitInfo;
+ if (split instanceof DataSplit) {
+ splitInfo = "snapshotId=" + ((DataSplit) split).snapshotId();
+ } else if (split instanceof IncrementalSplit) {
+ splitInfo = "snapshotId=" + ((IncrementalSplit)
split).snapshotId();
+ } else if (split instanceof ChainSplit) {
+ splitInfo = split.toString();
+ } else {
+ splitInfo = split.getClass().getSimpleName();
+ }
LOG.info(
- "LookupStreamingReader get splits from {} with snapshotId {}.",
+ "LookupStreamingReader get {} splits from {} ({}).",
+ splits.size(),
table.name(),
- snapshotId);
+ splitInfo);
}
@Nullable
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
index b00ac7e083..9964a8fa6c 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java
@@ -220,7 +220,7 @@ public class PrimaryKeyPartialLookupTable implements
LookupTable {
return new PrimaryKeyPartialLookupTable(
(filter, cacheRowFilter) ->
new LocalQueryExecutor(
- new LookupFileStoreTable(table, joinKey),
+ LookupFileStoreTable.create(table, joinKey),
projection,
tempPath,
filter,
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
index 7d7a053a16..6a9b2c8058 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkChainTableITCase.java
@@ -21,35 +21,50 @@ package org.apache.paimon.flink;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.flink.lookup.FullCacheLookupTable;
+import org.apache.paimon.flink.lookup.LookupFileStoreTable;
import org.apache.paimon.flink.sink.FlinkSinkBuilder;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.ChainTableFileStoreTable;
import org.apache.paimon.table.ChainTableStreamScan;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.ChainSplit;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DataTableScan;
+import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.utils.BlockingIterator;
+import org.apache.flink.api.common.JobStatus;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.ExternalizedCheckpointRetention;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.condition.EnabledIf;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -59,6 +74,11 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for chain table using Flink SQL. */
public class FlinkChainTableITCase extends CatalogITCaseBase {
+ @SuppressWarnings("unused")
+ static boolean isFlink2OrLater() {
+ return isFlinkVersionGreaterThanOrEqualTo("2.0");
+ }
+
private List<String> collectResult(String query) throws Exception {
List<String> result = new ArrayList<>();
try (CloseableIterator<Row> it = tEnv.executeSql(query).collect()) {
@@ -2301,8 +2321,6 @@ public class FlinkChainTableITCase extends
CatalogITCaseBase {
+ " VALUES (4, 1, 'delta_4')");
// Step 6: Collect Phase 2 output (should include delta@20250810)
- // BUG: If snapshot branch was not pinned, Phase 1 might have excluded
delta@20250808
- // after seeing snapshot@20250809, and Phase 2 would miss it too.
List<String> phase2 = collectRows(it, 1);
assertThat(phase2)
.as("Phase 2 should include new delta@20250810 data")
@@ -2378,8 +2396,6 @@ public class FlinkChainTableITCase extends
CatalogITCaseBase {
+ " VALUES (3, 1, 'delta_3')");
// Collect Phase 2 output
- // BUG: If Phase 2 read bypasses branch-aware logic, this might fail
or produce
- // wrong results when snapshot/delta schemas diverge.
List<String> phase2 = collectRows(it, 1);
assertThat(phase2)
.as("Phase 2 should correctly read delta@20250810 with
branch-aware logic")
@@ -2387,4 +2403,1403 @@ public class FlinkChainTableITCase extends
CatalogITCaseBase {
it.close();
}
+
+ @Test
+ public void testLookupJoin() throws Exception {
+ // Create chain table as dimension table
+ sql(
+ "CREATE TABLE chain_dim ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim");
+
+ // Write snapshot branch
+ sql(
+ "INSERT OVERWRITE `chain_dim$branch_snapshot` PARTITION (dt =
'20250808')"
+ + " VALUES (1, 1, 'snap_1'), (2, 1, 'snap_2'), (3, 1,
'snap_3')");
+
+ // Write delta branch (new partition, no anchor in snapshot)
+ sql(
+ "INSERT OVERWRITE `chain_dim$branch_delta` PARTITION (dt =
'20250809')"
+ + " VALUES (2, 2, 'delta_2_updated'), (4, 1,
'delta_4')");
+
+ // Create source table as Paimon table
+ sql(
+ "CREATE TABLE source_t ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+
+ // Insert source data
+ sql("INSERT INTO source_t VALUES (1), (2), (3), (4)");
+
+ // Lookup join on k (non-partition key)
+ List<String> result =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_t AS S "
+ + "LEFT JOIN chain_dim /*+
OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k");
+
+ // Verify lookup results.
+ // Source has 4 rows (id=1,2,3,4). Lightweight Phase 1 dim data:
+ // snapshot@20250808: k=1(snap_1), k=2(snap_2), k=3(snap_3)
+ // delta@20250809 (no anchor merge): k=2(delta_2_updated),
k=4(delta_4)
+ // k=1 matches id=1 (snapshot only), k=2 matches id=2 (snapshot +
delta = 2 rows),
+ // k=3 matches id=3 (snapshot only), k=4 matches id=4 (delta only) → 5
lookup matches.
+ assertThat(result).hasSize(5);
+ assertThat(result)
+ .anyMatch(r -> r.contains("snap_1"))
+ .anyMatch(r -> r.contains("snap_2"))
+ .anyMatch(r -> r.contains("delta_2_updated"))
+ .anyMatch(r -> r.contains("snap_3"))
+ .anyMatch(r -> r.contains("delta_4"));
+ }
+
+ @Test
+ @EnabledIf(
+ value = "isFlink2OrLater",
+ disabledReason = "Custom shuffle lookup join requires Flink 2.x.")
+ public void testLookupJoinWithBucketShuffle() throws Exception {
+ sql(
+ "CREATE TABLE chain_dim_shuffle ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_shuffle");
+
+ // Write snapshot branch
+ sql(
+ "INSERT OVERWRITE `chain_dim_shuffle$branch_snapshot`
PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 'snap_1'), (2, 1, 'snap_2'), (3, 1,
'snap_3')");
+
+ // Write delta branch (new partition, no anchor in snapshot)
+ sql(
+ "INSERT OVERWRITE `chain_dim_shuffle$branch_delta` PARTITION
(dt = '20250809')"
+ + " VALUES (2, 2, 'delta_2_updated'), (4, 1,
'delta_4')");
+
+ // Create source table
+ sql(
+ "CREATE TABLE source_shuffle ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_shuffle VALUES (1), (2), (3), (4)");
+
+ String query =
+ "SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ S.id,
D.k, D.v "
+ + "FROM source_shuffle AS S "
+ + "JOIN chain_dim_shuffle "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k";
+
+ // Verify the execution plan actually uses bucket shuffle partitioning.
+ // "shuffle=[true]" in the LookupJoin node proves the planner
recognized the LOOKUP hint
+ // and will use the BucketShufflePartitioner provided by Paimon's
BaseDataTableSource.
+
sEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM,
2);
+ String explainPlan = sEnv.explainSql(query);
+ assertThat(explainPlan)
+ .as("EXPLAIN plan should contain shuffle=[true] proving bucket
shuffle is active")
+ .contains("shuffle=[true]");
+
+ // Expected lightweight Phase 1 data (no anchor merge):
+ // snapshot@20250808: k=1(snap_1), k=2(snap_2), k=3(snap_3)
+ // delta@20250809 (no anchor merge): k=2(delta_2_updated),
k=4(delta_4)
+ // Source has 4 rows (id=1,2,3,4). k=1 matches id=1 (snapshot only),
+ // k=2 matches id=2 (snapshot + delta = 2 rows), k=3 matches id=3
(snapshot only),
+ // k=4 matches id=4 (delta only) → 5 lookup matches.
+ List<Row> result = streamSqlBlockIter(query).collect(5);
+
+ assertThat(result).hasSize(5);
+ assertThat(result)
+ .anyMatch(r -> r.toString().contains("snap_1"))
+ .anyMatch(r -> r.toString().contains("snap_2"))
+ .anyMatch(r -> r.toString().contains("delta_2_updated"))
+ .anyMatch(r -> r.toString().contains("snap_3"))
+ .anyMatch(r -> r.toString().contains("delta_4"));
+ }
+
+ @Test
+ public void testLookupJoinDeltaOnly() throws Exception {
+ // Chain table with only delta data (no snapshot)
+ sql(
+ "CREATE TABLE chain_dim_delta_only ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_delta_only");
+
+ // Write only delta (no snapshot)
+ sql(
+ "INSERT OVERWRITE `chain_dim_delta_only$branch_delta`
PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 'v1'), (2, 1, 'v2'), (3, 1, 'v3')");
+
+ // Verify batch read works first
+ List<String> batchResult =
+ collectResult("SELECT k, v FROM chain_dim_delta_only WHERE dt
= '20250808'");
+ assertThat(batchResult).as("Batch read of delta-only chain
table").hasSize(3);
+
+ sql(
+ "CREATE TABLE source_delta_only ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_delta_only VALUES (1), (2), (3)");
+
+ List<String> result =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_delta_only AS S "
+ + "LEFT JOIN chain_dim_delta_only "
+ + "/*+ OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k");
+
+ assertThat(result).hasSize(3);
+ assertThat(result)
+ .containsExactlyInAnyOrder("+I[1, 1, v1]", "+I[2, 2, v2]",
"+I[3, 3, v3]");
+ }
+
+ @Test
+ public void testLookupJoinSnapshotOnly() throws Exception {
+ // Chain table with only snapshot data (no delta)
+ sql(
+ "CREATE TABLE chain_dim_snap_only ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_snap_only");
+
+ // Write only snapshot (no delta)
+ sql(
+ "INSERT OVERWRITE `chain_dim_snap_only$branch_snapshot` "
+ + "PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 's1'), (2, 1, 's2')");
+
+ sql(
+ "CREATE TABLE source_snap_only ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_snap_only VALUES (1), (2)");
+
+ List<String> result =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_snap_only AS S "
+ + "LEFT JOIN chain_dim_snap_only "
+ + "/*+ OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k");
+
+ assertThat(result).hasSize(2);
+ assertThat(result).containsExactlyInAnyOrder("+I[1, 1, s1]", "+I[2, 2,
s2]");
+ }
+
+ @Test
+ public void testLookupJoinRejectsPartitionKeyInJoinCondition() throws
Exception {
+ sql(
+ "CREATE TABLE chain_dim_pk ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_pk");
+
+ sql(
+ "INSERT OVERWRITE `chain_dim_pk$branch_delta` PARTITION (dt =
'20250808')"
+ + " VALUES (1, 1, 'v1')");
+
+ sql(
+ "CREATE TABLE source_pk ("
+ + " id BIGINT,"
+ + " dt STRING,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_pk VALUES (1, '20250808')");
+
+ // Join on partition key dt should fail
+ assertThatThrownBy(
+ () ->
+ collectResult(
+ "SELECT S.id, D.k "
+ + "FROM source_pk AS S "
+ + "LEFT JOIN chain_dim_pk "
+ + "/*+ OPTIONS('lookup.cache'
= 'full') */ "
+ + "FOR SYSTEM_TIME AS OF
S.proc_time AS D "
+ + "ON S.dt = D.dt"))
+ .rootCause()
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("partition keys");
+ }
+
+ @Test
+ public void testLookupJoinRejectsInvalidCacheMode() throws Exception {
+ sql(
+ "CREATE TABLE chain_dim_mode ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_mode");
+
+ sql(
+ "INSERT OVERWRITE `chain_dim_mode$branch_delta` PARTITION (dt
= '20250808')"
+ + " VALUES (1, 1, 'v1')");
+
+ sql(
+ "CREATE TABLE source_mode ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_mode VALUES (1)");
+
+ // cache-mode=memory is not supported for chain tables
+ assertThatThrownBy(
+ () ->
+ collectResult(
+ "SELECT S.id, D.k "
+ + "FROM source_mode AS S "
+ + "LEFT JOIN chain_dim_mode "
+ + "/*+ OPTIONS('lookup.cache'
= 'memory') */ "
+ + "FOR SYSTEM_TIME AS OF
S.proc_time AS D "
+ + "ON S.id = D.k"))
+ .rootCause()
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("cache mode");
+ }
+
+ @Test
+ public void testLookupJoinRejectsJoinKeyContainingPartitionKey() throws
Exception {
+ // PK = (dt, k). If join keys = (dt, k) = PK, this triggers the AUTO +
PK==JK path
+ // in FileStoreLookupFunction.open(). The chain table validation
should reject it
+ // because dt is a partition key.
+ sql(
+ "CREATE TABLE chain_dim_pkjk ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_pkjk");
+
+ sql(
+ "INSERT OVERWRITE `chain_dim_pkjk$branch_delta` PARTITION (dt
= '20250808')"
+ + " VALUES (1, 1, 'v1')");
+
+ sql(
+ "CREATE TABLE source_pkjk ("
+ + " id BIGINT,"
+ + " dt STRING,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_pkjk VALUES (1, '20250808')");
+
+ // Join on both dt and k (= full PK) should fail because dt is a
partition key
+ assertThatThrownBy(
+ () ->
+ collectResult(
+ "SELECT S.id, D.v "
+ + "FROM source_pkjk AS S "
+ + "LEFT JOIN chain_dim_pkjk "
+ + "FOR SYSTEM_TIME AS OF
S.proc_time AS D "
+ + "ON S.dt = D.dt AND S.id =
D.k"))
+ .rootCause()
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("partition keys");
+ }
+
+ @Test
+ public void testLookupJoinWithAutoCacheMode() throws Exception {
+ // Same as testLookupJoinDeltaOnly but without explicit 'lookup.cache'
= 'full' hint.
+ // Default cache mode is AUTO. LookupFileStoreTable.create() validates
chain table
+ // constraints and FileStoreLookupFunction excludes chain tables from
the AUTO path,
+ // preventing PrimaryKeyPartialLookupTable from being used.
+ sql(
+ "CREATE TABLE chain_dim_auto ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_auto");
+
+ sql(
+ "INSERT OVERWRITE `chain_dim_auto$branch_delta` PARTITION (dt
= '20250808')"
+ + " VALUES (1, 1, 'v1'), (2, 1, 'v2')");
+
+ sql(
+ "CREATE TABLE source_auto ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_auto VALUES (1), (2)");
+
+ // No explicit lookup.cache hint — defaults to AUTO, should work via
AUTO→FULL conversion
+ List<String> result =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_auto AS S "
+ + "LEFT JOIN chain_dim_auto "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k");
+
+ assertThat(result).hasSize(2);
+ assertThat(result).containsExactlyInAnyOrder("+I[1, 1, v1]", "+I[2, 2,
v2]");
+ }
+
+ @Test
+ public void testChainTableStreamScanIncrementalRefresh() throws Exception {
+ // Tests the ChainTableStreamScan directly at the API level.
+ // Verifies: first plan() = bootstrap (chain-merged ChainSplits),
+ // second plan() = incremental (delta DataSplits with new
data).
+ sql(
+ "CREATE TABLE chain_dim_incr ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_incr");
+
+ // Write initial delta data
+ sql(
+ "INSERT OVERWRITE `chain_dim_incr$branch_delta` PARTITION (dt
= '20250808')"
+ + " VALUES (1, 1, 'v1'), (2, 1, 'v2')");
+
+ // Get the chain table and create a LookupFileStoreTable with
ChainTableStreamScan
+ FileStoreTable table = (FileStoreTable) paimonTable("chain_dim_incr");
+ LookupFileStoreTable lookupTable = LookupFileStoreTable.create(table,
Arrays.asList("k"));
+
+ ChainTableStreamScan scan = (ChainTableStreamScan)
lookupTable.newStreamScan();
+
+ // First plan() = bootstrap: should return ChainSplits with initial
data
+ List<Split> bootstrapSplits = scan.plan().splits();
+ assertThat(bootstrapSplits).isNotEmpty();
+ assertThat(bootstrapSplits.get(0))
+ .as("Bootstrap should produce ChainSplits")
+ .isInstanceOf(ChainSplit.class);
+
+ // Second plan() = incremental: should be empty (no new delta data)
+ List<Split> emptySplits = scan.plan().splits();
+ assertThat(emptySplits).as("No new data, should be empty").isEmpty();
+
+ // Write new delta data
+ sql(
+ "INSERT INTO `chain_dim_incr$branch_delta` PARTITION (dt =
'20250809')"
+ + " VALUES (3, 1, 'v3')");
+
+ // Third plan() = incremental: should return DataSplits with new data
+ List<Split> incrementalSplits = scan.plan().splits();
+ assertThat(incrementalSplits)
+ .as("Should have incremental splits after new delta data")
+ .isNotEmpty();
+ assertThat(incrementalSplits.get(0))
+ .as("Incremental should produce DataSplits")
+ .isInstanceOf(DataSplit.class);
+ }
+
+ @Test
+ public void testLookupScanCheckpointRestore() throws Exception {
+ // Tests checkpoint/restore behavior of ChainTableStreamScan at the
API level.
+ // Verifies:
+ // 1. Before bootstrap, checkpoint() returns null.
+ // 2. After bootstrap, checkpoint() returns the delta position.
+ // 3. restore(id) sets bootstrapDone=true and positions delta scan
correctly.
+ // 4. After restore, plan() only returns NEW data (not
already-consumed data).
+ sql(
+ "CREATE TABLE chain_dim_ckp ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_ckp");
+
+ // Write initial delta data
+ sql(
+ "INSERT OVERWRITE `chain_dim_ckp$branch_delta` PARTITION (dt =
'20250808')"
+ + " VALUES (1, 1, 'v1'), (2, 1, 'v2')");
+
+ FileStoreTable table = (FileStoreTable) paimonTable("chain_dim_ckp");
+ LookupFileStoreTable lookupTable = LookupFileStoreTable.create(table,
Arrays.asList("k"));
+
+ ChainTableStreamScan scan = (ChainTableStreamScan)
lookupTable.newStreamScan();
+
+ // Before bootstrap, checkpoint should be null
+ assertThat(scan.checkpoint()).as("Before bootstrap, checkpoint should
be null").isNull();
+
+ // Bootstrap
+ List<Split> bootstrapSplits = scan.plan().splits();
+ assertThat(bootstrapSplits).isNotEmpty();
+
+ // After bootstrap, checkpoint should capture the delta position
+ Long checkpointId = scan.checkpoint();
+ assertThat(checkpointId)
+ .as("After bootstrap, checkpoint should capture delta
position")
+ .isNotNull();
+
+ // No new data, plan() should be empty
+ assertThat(scan.plan().splits()).isEmpty();
+
+ // Write new delta data
+ sql(
+ "INSERT INTO `chain_dim_ckp$branch_delta` PARTITION (dt =
'20250809')"
+ + " VALUES (3, 1, 'v3')");
+
+ // Incremental plan() should return new data
+ List<Split> incrSplits1 = scan.plan().splits();
+ assertThat(incrSplits1).isNotEmpty();
+
+ // Checkpoint again
+ Long checkpointId2 = scan.checkpoint();
+ assertThat(checkpointId2).isNotNull();
+ assertThat(checkpointId2)
+ .as("Second checkpoint should be after the first")
+ .isGreaterThan(checkpointId);
+
+ // Simulate restore from first checkpoint
+ ChainTableStreamScan restoredScan = (ChainTableStreamScan)
lookupTable.newStreamScan();
+ restoredScan.restore(checkpointId);
+
+ // After restore, plan() should return data from checkpointId onwards
+ // (i.e., the data at dt=20250809 that was written after the first
checkpoint)
+ List<Split> restoredSplits = restoredScan.plan().splits();
+ assertThat(restoredSplits)
+ .as("Restored scan should return data after checkpoint
position")
+ .isNotEmpty();
+
+ // A fresh scan (no restore) should bootstrap and then be empty
+ ChainTableStreamScan freshScan = (ChainTableStreamScan)
lookupTable.newStreamScan();
+ freshScan.plan(); // bootstrap
+ assertThat(freshScan.plan().splits())
+ .as("Fresh scan after bootstrap should have no incremental
data")
+ .isEmpty();
+ }
+
+ @Test
+ public void testLookupJoinWithPredicatePushdown() throws Exception {
+ // Tests that a WHERE condition on the dimension table is correctly
pushed down
+ // and produces correct lookup results with chain-merged data.
+ sql(
+ "CREATE TABLE chain_dim_pred ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_pred");
+
+ // Write snapshot branch with 3 rows
+ sql(
+ "INSERT OVERWRITE `chain_dim_pred$branch_snapshot` PARTITION
(dt = '20250808')"
+ + " VALUES (1, 1, 'snap_1'), (2, 1, 'snap_2'), (3, 1,
'snap_3')");
+
+ // Write delta branch with updated row k=2 and new row k=4
+ sql(
+ "INSERT OVERWRITE `chain_dim_pred$branch_delta` PARTITION (dt
= '20250809')"
+ + " VALUES (2, 2, 'delta_2_updated'), (4, 1,
'delta_4')");
+
+ sql(
+ "CREATE TABLE source_pred ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_pred VALUES (1), (2), (3), (4)");
+
+ // Lookup join with a predicate on the dimension table's v column.
+ // The predicate v LIKE '%updated%' should be pushed down to the chain
table scan.
+ // Chain-merged data: k=1(snap_1), k=2(delta_2_updated), k=3(snap_3),
k=4(delta_4).
+ // Only k=2 has v containing 'updated'.
+ List<String> result =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_pred AS S "
+ + "LEFT JOIN chain_dim_pred "
+ + "/*+ OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k "
+ + "WHERE D.v LIKE '%updated%'");
+
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0)).contains("delta_2_updated");
+ }
+
+ @Test
+ public void testLookupRejectsIncompatibleDeltaBranchConfig() throws
Exception {
+ // Tests that creating a lookup table for a chain table with
partial-update merge engine
+ // on the delta branch is rejected. The SQL lookup join path is
covered by
+ // testLookupRejectsAggregateOnDeltaBranch, which verifies the same
validation
+ // through the SQL path (prepareBranchOptions preserves
branch-specific merge-engine
+ // during table.copy(TableSchema)).
+ sql(
+ "CREATE TABLE chain_dim_partial_cfg ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_partial_cfg");
+
+ // Alter delta branch to partial-update (unsupported for incremental
lookup).
+ // The incremental read path (createNoMergeReader) does not apply the
merge engine,
+ // which would cause partial rows to overwrite complete cached data.
+ sql(
+ "ALTER TABLE `chain_dim_partial_cfg$branch_delta` SET ("
+ + " 'merge-engine' = 'partial-update'"
+ + ")");
+
+ // Load the chain table directly from catalog (preserves branch
on-disk options).
+ // Table loading succeeds because batch reads work fine with
PARTIAL_UPDATE.
+ FileStoreTable table = (FileStoreTable)
paimonTable("chain_dim_partial_cfg");
+
+ // Creating a lookup table should fail because the incremental read
path
+ // (ChainTableStreamScan) does not support PARTIAL_UPDATE on the delta
branch.
+ assertThatThrownBy(
+ () ->
+
org.apache.paimon.flink.lookup.LookupFileStoreTable.create(
+ table, Collections.singletonList("k")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("merge engine");
+ }
+
+ @Test
+ public void testStreamingReadRejectsPartialUpdateOnDeltaBranch() throws
Exception {
+ // Tests that creating a streaming scan for a chain table with
partial-update merge engine
+ // on the delta branch is rejected.
+ sql(
+ "CREATE TABLE chain_stream_partial ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_stream_partial");
+
+ sql(
+ "ALTER TABLE `chain_stream_partial$branch_delta` SET ("
+ + " 'merge-engine' = 'partial-update'"
+ + ")");
+
+ FileStoreTable table = (FileStoreTable)
paimonTable("chain_stream_partial");
+
+ // Creating a streaming scan should fail because ChainTableStreamScan
does not support
+ // PARTIAL_UPDATE on the delta branch.
+ assertThatThrownBy(() -> table.newStreamScan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("merge engine");
+ }
+
+ @Test
+ public void testStreamingReadRejectsAggregateOnDeltaBranch() throws
Exception {
+ // Tests that creating a streaming scan for a chain table with
aggregation merge engine
+ // on the delta branch is rejected.
+ sql(
+ "CREATE TABLE chain_stream_agg ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v INT,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_stream_agg");
+
+ sql(
+ "ALTER TABLE `chain_stream_agg$branch_delta` SET ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.v.aggregate-function' = 'sum'"
+ + ")");
+
+ FileStoreTable table = (FileStoreTable)
paimonTable("chain_stream_agg");
+
+ // Creating a streaming scan should fail because ChainTableStreamScan
does not support
+ // AGGREGATE on the delta branch.
+ assertThatThrownBy(() -> table.newStreamScan())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("merge engine");
+ }
+
+ @Test
+ public void testLookupRejectsAggregateOnDeltaBranch() throws Exception {
+ // Tests that creating a lookup table for a chain table with
aggregation merge engine
+ // on the delta branch is rejected.
+ sql(
+ "CREATE TABLE chain_dim_agg_cfg ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v INT,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_agg_cfg");
+
+ sql(
+ "ALTER TABLE `chain_dim_agg_cfg$branch_delta` SET ("
+ + " 'merge-engine' = 'aggregation',"
+ + " 'fields.v.aggregate-function' = 'sum'"
+ + ")");
+
+ FileStoreTable table = (FileStoreTable)
paimonTable("chain_dim_agg_cfg");
+
+ // Creating a lookup table should fail because ChainTableStreamScan
does not support
+ // AGGREGATE on the delta branch.
+ assertThatThrownBy(
+ () ->
+
org.apache.paimon.flink.lookup.LookupFileStoreTable.create(
+ table, Collections.singletonList("k")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("merge engine");
+
+ // Also verify through SQL lookup join path
+ sql(
+ "INSERT OVERWRITE `chain_dim_agg_cfg$branch_snapshot`
PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 10)");
+ sql(
+ "INSERT OVERWRITE `chain_dim_agg_cfg$branch_delta` PARTITION
(dt = '20250809')"
+ + " VALUES (2, 2, 20)");
+
+ sql(
+ "CREATE TABLE source_agg_cfg ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_agg_cfg VALUES (1), (2)");
+
+ String query =
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_agg_cfg AS S "
+ + "LEFT JOIN chain_dim_agg_cfg "
+ + "/*+ OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k";
+
+ assertThatThrownBy(() -> collectResult(query))
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("merge engine");
+ }
+
+ /**
+ * Tests that chain table lookup join rejects unsupported scan modes and
consumer-id. The
+ * validation is inherited from {@link
ChainTableFileStoreTable#newStreamScan()}.
+ */
+ @Test
+ public void testLookupRejectsUnsupportedScanMode() throws Exception {
+ sql(
+ "CREATE TABLE chain_dim_scan_mode ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_scan_mode");
+
+ FileStoreTable table = paimonTable("chain_dim_scan_mode");
+
+ // scan.mode=latest should be rejected for chain table lookup
+ FileStoreTable tableLatest =
table.copy(Collections.singletonMap("scan.mode", "latest"));
+ assertThatThrownBy(
+ () ->
+
org.apache.paimon.flink.lookup.LookupFileStoreTable.create(
+ tableLatest,
Collections.singletonList("k"))
+ .newStreamScan())
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("scan.mode=latest")
+ .hasMessageContaining("Chain table streaming read does not
support");
+
+ // consumer-id should be rejected for chain table lookup
+ FileStoreTable tableWithConsumer =
+ table.copy(Collections.singletonMap("consumer-id",
"my-consumer"));
+ assertThatThrownBy(
+ () ->
+
org.apache.paimon.flink.lookup.LookupFileStoreTable.create(
+ tableWithConsumer,
Collections.singletonList("k"))
+ .newStreamScan())
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("consumer mode")
+ .hasMessageContaining("consumer-id='my-consumer'");
+ }
+
+ @Test
+ @Timeout(180)
+ public void testLookupJoinWithCompactDeltaMonitorMode() throws Exception {
+ // Main table uses partial-update + force-lookup so that
+ // supportCompactDiffStreamingReading returns true and lookupScanMode
becomes
+ // COMPACT_DELTA_MONITOR.
+ sql(
+ "CREATE TABLE chain_dim_cdm ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'partial-update',"
+ + " 'force-lookup' = 'true',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_dim_cdm");
+
+ // Delta branch must be deduplicate for chain table incremental read.
+ sql("ALTER TABLE `chain_dim_cdm$branch_delta` SET ('merge-engine' =
'deduplicate')");
+
+ // Write data to both branches.
+ sql(
+ "INSERT OVERWRITE `chain_dim_cdm$branch_snapshot` PARTITION
(dt = '20250808')"
+ + " VALUES (1, 1, 'snap_v1'), (2, 2, 'snap_v2')");
+ sql(
+ "INSERT INTO `chain_dim_cdm$branch_delta` PARTITION (dt =
'20250809')"
+ + " VALUES (1, 3, 'delta_v1'), (3, 4, 'delta_v3')");
+
+ // Verify that LookupFileStoreTable.newRead() delegates to
wrapped.newRead()
+ // (not LookupCompactDiffRead, which would cause ClassCastException on
ChainSplit).
+ FileStoreTable table = paimonTable("chain_dim_cdm");
+ org.apache.paimon.flink.lookup.LookupFileStoreTable lookupTable =
+ org.apache.paimon.flink.lookup.LookupFileStoreTable.create(
+ table, Collections.singletonList("k"));
+
assertThat(lookupTable.newRead()).isInstanceOf(table.newRead().getClass());
+
+ // Run the lookup join and verify results.
+ sql(
+ "CREATE TABLE source_cdm ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ('connector' = 'paimon')");
+ sql("INSERT INTO source_cdm VALUES (1), (2), (3)");
+
+ List<String> results =
+ collectResult(
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_cdm AS S "
+ + "LEFT JOIN chain_dim_cdm "
+ + "/*+ OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k");
+
+ // k=1: appears in both snapshot (dt=20250808) and delta (dt=20250809)
partitions.
+ // Chain table merges at partition level, not row level, so both rows
appear.
+ // k=2: only in snapshot partition.
+ // k=3: only in delta partition.
+ assertThat(results)
+ .containsExactlyInAnyOrder(
+ "+I[1, 1, snap_v1]",
+ "+I[1, 1, delta_v1]",
+ "+I[2, 2, snap_v2]",
+ "+I[3, 3, delta_v3]");
+ }
+
+ /**
+ * Verifies that ChainTableStreamScan correctly propagates bucket filters
to its internal scans.
+ */
+ @Test
+ public void testStreamingReadBucketFilter() throws Exception {
+ sql(
+ "CREATE TABLE chain_bucket_filter ("
+ + " k BIGINT, seq BIGINT, v STRING, dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+
+ String db = tEnv.getCurrentDatabase();
+ sql("CALL sys.create_branch('%s.chain_bucket_filter', 'snapshot')",
db);
+ sql("CALL sys.create_branch('%s.chain_bucket_filter', 'delta')", db);
+ for (String tbl :
+ new String[] {
+ "chain_bucket_filter",
+ "chain_bucket_filter$branch_snapshot",
+ "chain_bucket_filter$branch_delta"
+ }) {
+ sql(
+ "ALTER TABLE `%s` SET ("
+ + " 'scan.fallback-snapshot-branch' = 'snapshot',"
+ + " 'scan.fallback-delta-branch' = 'delta')",
+ tbl);
+ }
+
+ // Write main branch
+ sql(
+ "INSERT OVERWRITE chain_bucket_filter PARTITION (dt =
'20250810')"
+ + " VALUES (1, 1, 'v1')");
+
+ // Write delta data across many keys to guarantee both buckets are
populated
+ for (int i = 1; i <= 50; i++) {
+ sql(
+ String.format(
+ "INSERT INTO `chain_bucket_filter$branch_delta`"
+ + " PARTITION (dt = '%d') VALUES (%d, 1,
'v%d')",
+ 20250809 + (i % 5), i, i));
+ }
+
+ FileStoreTable table = paimonTable("chain_bucket_filter");
+
+ // Verify data spans both buckets via delta branch batch scan
+ FileStoreTable deltaTable =
+ (FileStoreTable)
paimonTable("chain_bucket_filter$branch_delta");
+ java.util.Set<Integer> deltaBuckets = new java.util.HashSet<>();
+ for (Split split : deltaTable.newScan().plan().splits()) {
+ if (split instanceof DataSplit) {
+ deltaBuckets.add(((DataSplit) split).bucket());
+ }
+ }
+ assertThat(deltaBuckets)
+ .as("Test requires data in both buckets")
+ .containsExactlyInAnyOrder(0, 1);
+
+ // Without bucket filter: Phase 1 should return splits from both
buckets
+ ChainTableStreamScan scanAll = (ChainTableStreamScan)
table.newStreamScan();
+ TableScan.Plan planAll = scanAll.plan();
+ java.util.Set<Integer> bucketsAll =
collectBucketsFromChainSplits(planAll);
+ assertThat(bucketsAll)
+ .as("Without filter, should have data from both buckets")
+ .containsExactlyInAnyOrder(0, 1);
+
+ // With bucket filter (only bucket 0): Phase 1 should only return
bucket 0
+ ChainTableStreamScan scanFiltered = (ChainTableStreamScan)
table.newStreamScan();
+ scanFiltered.withBucketFilter(b -> b == 0);
+ TableScan.Plan planFiltered = scanFiltered.plan();
+ java.util.Set<Integer> bucketsFiltered =
collectBucketsFromChainSplits(planFiltered);
+ assertThat(bucketsFiltered)
+ .as("Bucket filter should restrict Phase 1 to bucket 0 only")
+ .containsExactly(0);
+ }
+
+ private java.util.Set<Integer>
collectBucketsFromChainSplits(TableScan.Plan plan) {
+ java.util.Set<Integer> buckets = new java.util.HashSet<>();
+ for (Split split : plan.splits()) {
+ if (split instanceof ChainSplit) {
+ for (String path : ((ChainSplit)
split).fileBucketPathMapping().values()) {
+ if (path.contains("bucket-0")) {
+ buckets.add(0);
+ } else if (path.contains("bucket-1")) {
+ buckets.add(1);
+ }
+ }
+ }
+ }
+ return buckets;
+ }
+
+ /**
+ * Verifies that lookup join on a chain table branch (e.g.,
t$branch_delta) works correctly.
+ * Branch tables may still carry chain-table.enabled=true, but they are
not wrapped in
+ * FallbackReadFileStoreTable, so they should be treated as regular tables
for lookup join.
+ */
+ @Test
+ public void testLookupJoinOnBranchTable() throws Exception {
+ sql(
+ "CREATE TABLE chain_branch_lookup ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches("chain_branch_lookup");
+
+ // Write some data to delta branch
+ sql(
+ "INSERT OVERWRITE `chain_branch_lookup$branch_delta` PARTITION
(dt = '20250808')"
+ + " VALUES (1, 1, 'v1'), (2, 1, 'v2')");
+
+ // Create source table
+ sql(
+ "CREATE TABLE source_branch ("
+ + " id BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+ sql("INSERT INTO source_branch VALUES (1), (2)");
+
+ // Lookup join on branch table should work correctly
+ String query =
+ "SELECT S.id, D.k, D.v "
+ + "FROM source_branch AS S "
+ + "LEFT JOIN `chain_branch_lookup$branch_delta` "
+ + "/*+ OPTIONS('lookup.cache' = 'full') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k";
+
+ List<String> result = collectResult(query);
+ assertThat(result)
+ .as("Lookup join on branch table should return matching rows")
+ .hasSize(2)
+ .containsExactlyInAnyOrder("+I[1, 1, v1]", "+I[2, 2, v2]");
+ }
+
+ /**
+ * Tests that chain table lookup join refresh works correctly for both
async and non-async
+ * modes. This test verifies the refresh logic in {@code
FullCacheLookupTable.refresh()} which
+ * has different code paths for async vs non-async refresh. For chain
tables, the async path
+ * skips the backlog calculation (since outer table and delta branch use
different snapshot
+ * sequences).
+ *
+ * <p>The lookup join job is started BEFORE inserting source data, so the
lookup cache is warmed
+ * up before any source data arrives.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @Timeout(120)
+ public void testLookupJoinRefresh(boolean asyncRefresh) throws Exception {
+ String tableName = "chain_refresh_" + (asyncRefresh ? "async" :
"sync");
+
+ tEnv.useCatalog("PAIMON");
+ tEnv.useDatabase("default");
+
+ sql(
+ "CREATE TABLE "
+ + tableName
+ + " ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches(tableName);
+
+ sql(
+ "INSERT INTO `"
+ + tableName
+ + "$branch_snapshot` PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 'snap_1'), (2, 1, 'snap_2')");
+
+ sql(
+ "INSERT INTO `"
+ + tableName
+ + "$branch_delta` PARTITION (dt = '20250809')"
+ + " VALUES (3, 1, 'delta_3')");
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.enableCheckpointing(100);
+ env.setParallelism(1);
+
+ EnvironmentSettings settings =
EnvironmentSettings.newInstance().inStreamingMode().build();
+ StreamTableEnvironment streamTableEnv =
StreamTableEnvironment.create(env, settings);
+
+ streamTableEnv.registerCatalog("PAIMON",
tEnv.getCatalog("PAIMON").get());
+ streamTableEnv.useCatalog("PAIMON");
+ streamTableEnv.useDatabase("default");
+
+ streamTableEnv.executeSql(
+ "CREATE TABLE IF NOT EXISTS source_refresh ("
+ + " id BIGINT,"
+ + " v BIGINT,"
+ + " proc_time AS PROCTIME()"
+ + ") WITH ("
+ + " 'connector' = 'paimon'"
+ + ")");
+
+ streamTableEnv.executeSql(
+ "CREATE TABLE IF NOT EXISTS sink_refresh ("
+ + " id BIGINT,"
+ + " k BIGINT,"
+ + " v STRING"
+ + ") WITH ("
+ + " 'connector' = 'paimon',"
+ + " 'primary-key' = 'id',"
+ + " 'bucket' = '1'"
+ + ")");
+
+ streamTableEnv
+ .executeSql("INSERT INTO source_refresh VALUES (1, 1), (2, 2),
(3, 3)")
+ .await();
+
+ // Submit lookup join job BEFORE inserting source data
+ String query =
+ String.format(
+ "INSERT INTO sink_refresh "
+ + "SELECT S.id, D.k, D.v "
+ + "FROM source_refresh AS S "
+ + "LEFT JOIN "
+ + tableName
+ + " /*+ OPTIONS('lookup.cache' = 'full',
'lookup.refresh-async' = '%s', 'continuous.refresh-interval' = '1s') */ "
+ + "FOR SYSTEM_TIME AS OF S.proc_time AS D "
+ + "ON S.id = D.k",
+ asyncRefresh);
+
+ TableResult tableResult = streamTableEnv.executeSql(query);
+ JobClient jobClient =
+ tableResult
+ .getJobClient()
+ .orElseThrow(() -> new RuntimeException("Failed to get
JobClient"));
+
+ try {
+ waitForJobRunning(jobClient);
+
+ waitForQueryResult(
+ "SELECT * FROM sink_refresh ORDER BY id", results ->
results.size() == 3);
+
+ // Insert new delta data
+ sql(
+ "INSERT INTO `"
+ + tableName
+ + "$branch_delta` PARTITION (dt = '20250810')
VALUES (4, 1, 'delta_4')");
+
+ waitForQueryResult(
+ "SELECT * FROM " + tableName + " WHERE dt = '20250810'
ORDER BY k",
+ results -> results.size() == 4);
+
+ long startTime = System.currentTimeMillis();
+ long value = 0;
+ List<String> results = null;
+ while (System.currentTimeMillis() - startTime < 30000L) {
+ // Insert new source data to trigger lookup
+ streamTableEnv
+ .executeSql("INSERT INTO source_refresh VALUES (4, " +
value + ")")
+ .await();
+
+ results = collectResult("SELECT * FROM sink_refresh WHERE id =
4");
+ if (results.size() == 1 && results.get(0).contains("delta_4"))
{
+ break;
+ }
+ Thread.sleep(500);
+ }
+
+ assertThat(results).hasSize(1).contains("+I[4, 4, delta_4]");
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ /**
+ * Tests that {@link FullCacheLookupTable#refresh()} uses the delta
branch's snapshot manager
+ * for chain tables, so that async refresh is chosen when the delta
backlog is small.
+ */
+ @Test
+ @Timeout(120)
+ public void testChainTableLookupRefreshAsyncPath() throws Exception {
+ String tableName = "chain_refresh_async_path";
+
+ tEnv.useCatalog("PAIMON");
+ tEnv.useDatabase("default");
+
+ sql(
+ "CREATE TABLE "
+ + tableName
+ + " ("
+ + " k BIGINT,"
+ + " seq BIGINT,"
+ + " v STRING,"
+ + " dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'primary-key' = 'dt,k',"
+ + " 'bucket-key' = 'k',"
+ + " 'bucket' = '2',"
+ + " 'sequence.field' = 'seq',"
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'chain-table.enabled' = 'true',"
+ + " 'partition.timestamp-pattern' = '$dt',"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd'"
+ + ")");
+ setupChainTableBranches(tableName);
+
+ // Write initial data to delta branch so the lookup table can
bootstrap.
+ sql(
+ "INSERT OVERWRITE `"
+ + tableName
+ + "$branch_delta` PARTITION (dt = '20250808')"
+ + " VALUES (1, 1, 'delta_1')");
+
+ // Get the chain table and enable async refresh with a
pending-snapshot-count of 0.
+ FileStoreTable table = paimonTable(tableName);
+ Map<String, String> lookupOptions = new HashMap<>();
+ lookupOptions.put("lookup.refresh.async", "true");
+ lookupOptions.put("lookup.refresh.async.pending-snapshot-count", "0");
+ table = table.copy(lookupOptions);
+
+ // Create and open a FullCacheLookupTable directly so we can inspect
refreshFuture.
+ File tempDir = new File(temporaryFolder.toFile(), tableName);
+ tempDir.mkdirs();
+ FullCacheLookupTable.Context context =
+ new FullCacheLookupTable.Context(
+ table,
+ new int[] {0, 1, 2, 3},
+ null,
+ null,
+ tempDir,
+ Collections.singletonList("k"),
+ null);
+ FullCacheLookupTable lookupTable =
FullCacheLookupTable.create(context, 0);
+ lookupTable.open();
+
+ try {
+ // Write many commits to the MAIN table. The main table and delta
branch maintain
+ // independent snapshot sequences, so this inflates the main
table's snapshot id
+ // while the delta branch remains at a low snapshot id.
+ for (int i = 0; i < 10; i++) {
+ sql(
+ "INSERT INTO "
+ + tableName
+ + " PARTITION (dt = '20250810')"
+ + " VALUES ("
+ + (100 + i)
+ + ", "
+ + i
+ + ", 'main_"
+ + i
+ + "')");
+ }
+
+ // Write a single new delta record so the delta branch has a small
backlog.
+ sql(
+ "INSERT INTO `"
+ + tableName
+ + "$branch_delta` PARTITION (dt = '20250810')"
+ + " VALUES (3, 1, 'delta_3')");
+
+ // The delta branch has only one new snapshot, so the async
refresh path should be
+ // chosen.
+ lookupTable.refresh();
+
+ assertThat(lookupTable.getRefreshFuture())
+ .as(
+ "Chain table lookup refresh should use the async
path when the delta backlog is small.")
+ .isNotNull();
+
+ // Wait for the async refresh to complete before closing.
+ lookupTable.getRefreshFuture().get();
+ } finally {
+ lookupTable.close();
+ }
+ }
+
+ /** Helper method: poll a query until the result matches the expected
condition or timeout. */
+ private void waitForQueryResult(
+ String query, java.util.function.Predicate<List<String>>
condition) throws Exception {
+ long startTime = System.currentTimeMillis();
+ while (System.currentTimeMillis() - startTime < 30000L) {
+ List<String> results = collectResult(query);
+ if (condition.test(results)) {
+ return;
+ }
+ Thread.sleep(500);
+ }
+ throw new RuntimeException("Timed out waiting for query result: " +
query);
+ }
+
+ /** Helper method: wait for job to reach target status. */
+ private void waitForJobRunning(JobClient jobClient) throws Exception {
+ long startTime = System.currentTimeMillis();
+ JobStatus currentStatus = null;
+ while (System.currentTimeMillis() - startTime < (long) 30000) {
+ CompletableFuture<JobStatus> statusFuture =
jobClient.getJobStatus();
+ currentStatus = statusFuture.get();
+
+ if (currentStatus == JobStatus.RUNNING) {
+ return;
+ }
+
+ if (currentStatus.isGloballyTerminalState()) {
+ throw new RuntimeException(
+ "Job terminated unexpectedly with status: " +
currentStatus);
+ }
+
+ Thread.sleep(500);
+ }
+ throw new RuntimeException(
+ "Timed out waiting for job status running. Current status: " +
currentStatus);
+ }
}
