[core][flink] Support anchor-based merge in starting phase for chain table streaming read

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

@@ -222,9 +222,11 @@ you will get the following result:
 
 Chain tables support Flink streaming read. A streaming read job operates in two phases:
 
-1. **Full load phase**: Produces a full result by reading the latest snapshot partition (per group)
-   and delta partitions that come after it. For each partition group, only the most recent snapshot
-   partition is included — older snapshot partitions are considered outdated and excluded.
+1. **Full load phase**: By default it produces a lightweight result by reading the latest
+   snapshot partition (per group) and delta partitions that come after it. Older snapshot
+   partitions are excluded. You can enable `chain-table.streaming.merge-snapshot` to perform
+   anchor-based chain merging in this phase, allowing cross-branch `DELETE` records to be
+   resolved together with the snapshot data.
 2. **Incremental phase**: Continuously reads new commits from the delta branch as they arrive.
 
 ### Write-Side Requirements
@@ -251,6 +253,28 @@ SET 'execution.runtime-mode' = 'streaming';
 INSERT INTO downstream_sink SELECT * FROM default.t;
 ```
 
+### Merge Snapshot in Full Load Phase
+
+By default, the full-load phase is lightweight: for each group it reads the latest snapshot and later
+delta partitions as separate splits. This is fast but cross-branch deletes are invisible — the `DELETE`
+records in the delta branch cannot be deleted in the snapshot branch.
+
+If you need a fully reconciled starting snapshot, enable merge mode:
+
+```sql
+ALTER TABLE default.t SET (
+  'chain-table.streaming.merge-snapshot' = 'true'
+);
+```
+
+With merge mode enabled, the full-load phase merges the latest snapshot partition per group with
+delta partitions whose chain key is strictly greater than the snapshot's, so cross-branch
+deletes are correctly resolved. The trade-off is a heavier startup scan.
+
+To reduce the overhead, run `CALL sys.compact_chain_table(...)` periodically.
+After compaction, only the delta changes that arrived after compaction need to be merged.
+
+
 ### Limitations
 
 - The incremental phase only monitors the **delta branch**. Writes to the snapshot branch are
@@ -268,6 +292,12 @@ INSERT INTO downstream_sink SELECT * FROM default.t;
   specific partition, use batch mode instead.
 - The delta branch must use the `DEDUPLICATE` merge engine (default). Other merge engine
   types are not supported.
+- The `changelog-producer` option must be `none` (default) or `input`; `lookup` and `full-compaction`
+  are not supported for chain tables.
+  - When `changelog-producer` is `none`, Flink's operator normalizes records by the full
+    primary key including the chain partition. The records of `-D`/`-U` in a different chain partition
+    than the original records of `+I` will be dropped. Use `input` if downstream must receive cross-partition
+    changelog records.
 
 ## Lookup Join
 

docs/generated/core_configuration.html

@@ -158,6 +158,12 @@
             <td>Boolean</td>
             <td>Whether enabled chain table.</td>
         </tr>
+        <tr>
+            <td><h5>chain-table.streaming.merge-snapshot</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If true, the starting phase of chain table streaming read performs anchor-based chain merging: for each group it merges the latest snapshot partition with delta partitions whose chain key is strictly greater than the snapshot chain key. This allows streaming readers to see cross-branch deletions and updates at the cost of a heavier startup scan. When false (default), the starting phase only reads the latest snapshot partition per group and later delta partitions as separate splits, which is lightweight but may not reflect cross-branch deletes.</td>
+        </tr>
         <tr>
             <td><h5>changelog-file.compression</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -291,6 +291,22 @@ public InlineElement getDescription() {
                                     + "suffix of the table's partition keys. Comma-separated. "
                                     + "If not set, all partition keys participate in chain.");
 
+    public static final ConfigOption<Boolean> CHAIN_TABLE_STREAMING_MERGE_SNAPSHOT =
+            key("chain-table.streaming.merge-snapshot")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, the starting phase of chain table streaming read performs "
+                                    + "anchor-based chain merging: for each group it merges the "
+                                    + "latest snapshot partition with delta partitions whose chain "
+                                    + "key is strictly greater than the snapshot chain key. This "
+                                    + "allows streaming readers to see cross-branch deletions and "
+                                    + "updates at the cost of a heavier startup scan. When false "
+                                    + "(default), the starting phase only reads the latest snapshot "
+                                    + "partition per group and later delta partitions as separate "
+                                    + "splits, which is lightweight but may not reflect cross-branch "
+                                    + "deletes.");
+
     public static final String FILE_FORMAT_ORC = "orc";
     public static final String FILE_FORMAT_AVRO = "avro";
     public static final String FILE_FORMAT_PARQUET = "parquet";
@@ -4148,6 +4164,10 @@ public List<String> chainTableChainPartitionKeys() {
         return Arrays.stream(value.split(",")).map(String::trim).collect(Collectors.toList());
     }
 
+    public boolean chainTableStreamingMergeSnapshot() {
+        return options.get(CHAIN_TABLE_STREAMING_MERGE_SNAPSHOT);
+    }
+
     public boolean formatTableImplementationIsPaimon() {
         return options.get(FORMAT_TABLE_IMPLEMENTATION) == FormatTableImplementation.PAIMON;
     }

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

@@ -59,7 +59,6 @@
 import java.util.stream.Collectors;
 
 import static org.apache.paimon.utils.Preconditions.checkArgument;
-import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
 /**
  * Chain table which mainly read from the snapshot branch. However, if the snapshot branch does not
@@ -343,7 +342,6 @@ public Plan plan() {
                 for (List<BinaryRow> deltaPartitionsInGroup : groupedDeltaPartitions.values()) {
 
                     // Sort delta by chain dimension ascending.
-                    // chainPartitionForCompare avoids copying BinaryRow in the comparator hot path.
                     deltaPartitionsInGroup.sort(
                             (a, b) ->
                                     chainPartitionComparator.compare(
@@ -415,69 +413,26 @@ public Plan plan() {
                             deltaScan.withPartitionFilter(selectedDeltaPartitions);
                         }
 
-                        List<Split> subSplits = deltaScan.plan().splits();
-                        Set<String> snapshotFileNames = new HashSet<>();
+                        List<DataSplit> deltaSubSplits =
+                                deltaScan.plan().splits().stream()
+                                        .map(s -> (DataSplit) s)
+                                        .collect(Collectors.toList());
+                        List<DataSplit> snapshotSubSplits = new ArrayList<>();
                         if (partitionPairs.getValue() != null) {
                             snapshotScan.withPartitionFilter(
                                     Collections.singletonList(partitionPairs.getValue()));
-                            List<Split> mainSubSplits = snapshotScan.plan().splits();
-                            snapshotFileNames =
-                                    mainSubSplits.stream()
-                                            .flatMap(
-                                                    s ->
-                                                            ((DataSplit) s)
-                                                                    .dataFiles().stream()
-                                                                            .map(
-                                                                                    DataFileMeta
-                                                                                            ::fileName))
-                                            .collect(Collectors.toSet());
-                            subSplits.addAll(mainSubSplits);
-                        }
-                        Map<Integer, List<DataSplit>> bucketSplits = new LinkedHashMap<>();
-                        Integer bucketInAll = null;
-                        for (Split split : subSplits) {
-                            DataSplit dataSplit = (DataSplit) split;
-                            Integer totalBuckets = dataSplit.totalBuckets();
-                            checkNotNull(totalBuckets);
-                            if (bucketInAll == null) {
-                                bucketInAll = totalBuckets;
-                            } else {
-                                checkArgument(
-                                        totalBuckets.equals(bucketInAll),
-                                        "Inconsistent bucket num " + dataSplit.bucket());
-                            }
-
-                            bucketSplits
-                                    .computeIfAbsent(dataSplit.bucket(), k -> new ArrayList<>())
-                                    .add(dataSplit);
-                        }
-                        for (Map.Entry<Integer, List<DataSplit>> entry : bucketSplits.entrySet()) {
-                            HashMap<String, String> fileBucketPathMapping = new HashMap<>();
-                            HashMap<String, String> fileBranchMapping = new HashMap<>();
-                            List<DataSplit> splitList = entry.getValue();
-                            for (DataSplit dataSplit : splitList) {
-                                for (DataFileMeta file : dataSplit.dataFiles()) {
-                                    fileBucketPathMapping.put(
-                                            file.fileName(), dataSplit.bucketPath());
-                                    String branch =
-                                            snapshotFileNames.contains(file.fileName())
-                                                    ? options.scanFallbackSnapshotBranch()
-                                                    : options.scanFallbackDeltaBranch();
-                                    fileBranchMapping.put(file.fileName(), branch);
-                                }
-                            }
-                            ChainSplit split =
-                                    new ChainSplit(
-                                            partitionPairs.getKey(),
-                                            entry.getValue().stream()
-                                                    .flatMap(
-                                                            dataSplit ->
-                                                                    dataSplit.dataFiles().stream())
-                                                    .collect(Collectors.toList()),
-                                            fileBranchMapping,
-                                            fileBucketPathMapping);
-                            splits.add(split);
+                            snapshotSubSplits =
+                                    snapshotScan.plan().splits().stream()
+                                            .map(s -> (DataSplit) s)
+                                            .collect(Collectors.toList());
                         }
+                        splits.addAll(
+                                ChainTableUtils.buildChainSplits(
+                                        partitionPairs.getKey(),
+                                        snapshotSubSplits,
+                                        deltaSubSplits,
+                                        options.scanFallbackSnapshotBranch(),
+                                        options.scanFallbackDeltaBranch()));
                     }
                 }
             }