[flink] Support lookup join for Chain Table

[yunfengzhou-hub]

Purpose

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.

Bug fixes in existing paths

• ChainTableStreamScan.withBucketFilter(): Bucket filter was not propagated to inner scans nor stored for replay on local scans created in planStarting(). This caused bucket filter to be silently dropped during streaming read.
• ChainGroupReadTable.copy() methods: Branch options were not properly prepared — scan.mode from the parent table could propagate to branch tables (conflicting with explicit snapshot pinning), and bucket values could be lost during copy(TableSchema). Added prepareBranchOptions() to consistently handle branch name, bucket restoration, and chain-table-level directive stripping across all copy methods.
• ChainTableUtils: Added validateChainTableForIncrementalRead() to centralize validation that the delta branch uses DEDUPLICATE merge engine for incremental read paths (streaming read and lookup join).
• ChainKeyValueFileReaderFactory.getDataSchema(): Missing return statement caused fall-through to incorrect branch logic when file belongs to the current branch.

New lookup join capabilities

• LookupFileStoreTable: Added chain table awareness. When the wrapped table is a chain table, newStreamScan() returns ChainTableStreamScan. Added validation to reject partition keys in join condition and unsupported cache modes.
• ChainTableStreamScan: Added pinnedOptions() helper to set both scan.snapshot-id and scan.mode=from-snapshot when pinning branch tables, preventing conflicts with scan.mode=latest-full inherited from the lookup join path.
• ChainSplit: Added toString() for debugging.
• Documentation: Added Lookup Join section to chain table docs.

Tests

Added comprehensive test coverage in FlinkChainTableITCase, including basic lookup join, bucket shuffle, delta-only/snapshot-only scenarios, constraint validation, incremental refresh, checkpoint/restore, predicate pushdown, and merge engine rejection.

[JingsongLi]

This still loses branch-specific table options on the SQL lookup path. BaseDataTableSource.timeTravelDisabledTable() builds a schema from the outer chain table options and calls table.copy(newSchema) before LookupFileStoreTable.create() validates the chain table. In copy(TableSchema), this helper starts from newTableSchema.options() and only restores branch and bucket, so a delta branch altered to merge-engine=partial-update or aggregation is copied back as the main table merge engine. The new validation then sees DEDUPLICATE and the unsupported delta branch is accepted. Please preserve branch-owned options from sourceTable.schema().options() (then overlay only safe dynamic scan options), and add a SQL lookup test for partial-update / aggregation delta branches.

[JingsongLi]

For chain-table lookup this scan checkpoints the delta branch position, but the existing full-cache refresh logic still compares reader.nextSnapshotId() with context.table.snapshotManager().latestSnapshotId() / the outer table latest snapshot. Those are different snapshot sequences for chain tables. With lookup.refresh.async or lookup.refresh.full-load-threshold, the backlog calculation can be wrong and may skip the intended sync/full refresh behavior. Please either make the lookup refresh path use the delta branch snapshot manager for chain tables, or explicitly disable those backlog/full-load threshold checks for chain-table lookup.

[JingsongLi]

This check can misclassify branch tables. t$branch_delta / t$branch_snapshot may still carry chain-table.enabled=true, but FileStoreTableFactory#createChainTable returns the branch table directly when the current branch is the snapshot or delta branch, so it is not a FallbackReadFileStoreTable. A lookup join on a branch table would hit this cast and fail with ClassCastException (and the AUTO partial-lookup path is also skipped). Please gate this on the actual wrapper shape, e.g. table instanceof FallbackReadFileStoreTable && ((FallbackReadFileStoreTable) table).other() instanceof ChainGroupReadTable, and add a branch-table lookup regression test.

[JingsongLi]

Good catch for the async backlog path, but the same cross-branch snapshot mismatch still exists in FileStoreLookupFunction#shouldDoFullLoad: it compares the outer table latest snapshot id with lookupTable.nextSnapshotId(), which for chain tables is the delta-branch checkpoint. If users enable lookup.refresh.full-load-threshold, this can trigger unnecessary full reloads or miss a real delta backlog. Please apply the same chain-table guard there, or compare against the delta branch's latest snapshot id instead.

[JingsongLi]

This does not really verify incremental refresh on the same lookup table. The second collectResult(query) starts a new query/job and rebuilds the lookup cache from scratch, so it can pass even if FullCacheLookupTable.refresh() never picks up the new delta snapshot. Could this be changed to keep one lookup table/job alive across the delta write, or to test the lookup function/table directly so the same cache instance performs the refresh?

[JingsongLi]

This still does not hit the chain-table path in full-cache refresh. this.table is assigned from context.table, and Context wraps the original table with LookupFileStoreTable.create(...), so the object here is a LookupFileStoreTable, not a FallbackReadFileStoreTable. As a result isChainTable is always false and async refresh still compares reader.nextSnapshotId() from the delta-branch ChainTableStreamScan with the outer/lookup table snapshot manager. Please unwrap LookupFileStoreTable.wrapped() (or keep the original table in the context) before checking for FallbackReadFileStoreTable, and add a refresh test that fails if this branch is not taken.

[JingsongLi]

Thanks for the update. The main chain-table lookup path looks good to me now, and the new tests cover the previous branch-option and delta-snapshot refresh issues.

I found one remaining concern: LookupFileStoreTable.newStreamScan() creates new ChainTableStreamScan(chainGroupReadTable) directly for chain-table lookup, which bypasses the validation in ChainTableFileStoreTable.newStreamScan() for unsupported scan.mode and consumer-id. As a result, a lookup join on a chain table may silently ignore settings such as scan.mode=compacted-full or consumer-id, while normal chain-table streaming would reject them. Could we either reuse wrapped.newStreamScan() here, or apply the same validation before creating ChainTableStreamScan? It would also be good to add a regression test for lookup join with unsupported scan.mode / consumer-id.

Minor doc suggestion: the lookup-join limitations should also mention that partition filters / scan.partitions are not supported for chain-table lookup, otherwise users may think only partition keys in the join condition are forbidden.

[JingsongLi]

I found one remaining issue in the chain-table lookup read path.

LookupFileStoreTable.newStreamScan() now correctly delegates chain-table scans to the wrapped table, so the bootstrap phase can produce ChainSplits from ChainTableStreamScan. However, LookupFileStoreTable.newRead() still selects the reader based on lookupScanMode. If the outer/main chain table options make TableScanUtils.supportCompactDiffStreamingReading(table) return true, lookupScanMode becomes COMPACT_DELTA_MONITOR and newRead() returns LookupCompactDiffRead.

That reader only handles DataSplit as the full-load path; for any other split it delegates to IncrementalCompactDiffSplitRead, which casts the split to IncrementalSplit. Therefore a bootstrap ChainSplit from the chain-table scan can fail with a ClassCastException.

Could we make chain-table lookup always delegate newRead() to wrapped.newRead() (or otherwise prevent chain-table lookup from using COMPACT_DELTA_MONITOR), and add a regression test for a chain table whose outer/main table options would select compact-diff lookup mode while the delta branch remains DEDUPLICATE?

[JingsongLi]

+1