[flink] Support stream read Chain Table

Author: yunfengzhou-hub

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

@@ -129,7 +129,6 @@ ALTER TABLE `default`.`t$branch_delta` SET (
 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.
-- Both Spark and Flink batch read/write are supported. Flink streaming read/write is not supported.
 - Chain compact is not supported for now, and it will be supported later.
 - Deletion vector is not supported for chain table.
 
@@ -191,6 +190,49 @@ you will get the following result:
 +---+----+-----+ 
 ```
 
+## Streaming Read
+
+Chain tables support Flink streaming read. A streaming read job first produces a full chain-merge
+result (same as the batch [Hybrid Query](#hybrid-query) above), then continuously reads new data
+from the delta branch as it arrives.
+
+### Write-Side Requirements
+
+Streaming read assumes the chain table follows the standard write pattern described at the top of
+this page:
+
+- **Snapshot branch** receives periodic full data (e.g., a daily ODS binlog dump job writes via
+  `INSERT OVERWRITE t$branch_snapshot`). Each snapshot partition represents a complete view of the
+  data at that point in time.
+- **Delta branch** receives incremental changes between snapshots (e.g., a batch job writes the
+  current day's new/updated records via `INSERT INTO t$branch_delta`). Each delta partition
+  contains only the changes for that period. The delta branch must have a
+  [changelog producer](./changelog-producer) configured (e.g., `'changelog-producer' = 'input'`)
+  for streaming read to work.
+
+The streaming read relies on this pattern to produce correct results. The full load phase uses the
+nearest snapshot as an anchor and merges forward through delta partitions using the chain merge
+strategy. After the full load, only new delta branch commits are picked up — writes to the
+snapshot branch do not trigger streaming output. To incorporate a new snapshot, restart the
+streaming job.
+
+### Usage
+
+```sql
+SET 'execution.runtime-mode' = 'streaming';
+
+INSERT INTO downstream_sink SELECT * FROM default.t;
+```
+
+### Limitations
+
+- The incremental phase only monitors the **delta branch**. Writes to the snapshot branch are
+  not detected until the streaming job is restarted.
+- The chain-table-aware streaming 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` with existing progress —
+  the standard streaming scan is used instead, bypassing the chain merge logic.
+
 ## Group Partition
 
 In real-world scenarios, a table often has multiple partition dimensions. For example, data may be

paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java

@@ -248,9 +248,10 @@ public RecordReader<KeyValue> createReader(DataSplit split) throws IOException {
 
     public RecordReader<KeyValue> createChainReader(ChainSplit chainSplit) throws IOException {
         List<DataFileMeta> files = chainSplit.dataFiles();
+        BinaryRow logicalPartition = chainSplit.logicalPartition();
         ChainReadContext chainReadContext =
                 new ChainReadContext.Builder()
-                        .withLogicalPartition(chainSplit.logicalPartition())
+                        .withLogicalPartition(logicalPartition)
                         .withFileBranchPathMapping(chainSplit.fileBranchMapping())
                         .withFileBucketPathMapping(chainSplit.fileBucketPathMapping())
                         .build();

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

@@ -233,6 +233,27 @@ public ChainTableBatchScan withBucketFilter(Filter<Integer> bucketFilter) {
          *       view of the queried partition.
          * </ol>
          */
+
+        /**
+         * Returns all {@link DataSplit}s from the delta (fallback) branch. Used by {@link
+         * ChainTableStreamScan} to read delta data directly with original partition values.
+         */
+        public List<DataSplit> getDeltaSplits() {
+            return fallbackScan.plan().splits().stream()
+                    .map(s -> (DataSplit) s)
+                    .collect(Collectors.toList());
+        }
+
+        /**
+         * Returns all {@link DataSplit}s from the snapshot (main) branch. Used by {@link
+         * ChainTableStreamScan} to read snapshot-only partitions.
+         */
+        public List<DataSplit> getSnapshotSplits() {
+            return mainScan.plan().splits().stream()
+                    .map(s -> (DataSplit) s)
+                    .collect(Collectors.toList());
+        }
+
         @Override
         public Plan plan() {
             List<Split> splits = new ArrayList<>();
@@ -520,8 +541,15 @@ public TableRead withIOManager(IOManager ioManager) {
 
         @Override
         public RecordReader<InternalRow> createReader(Split split) throws IOException {
-            checkArgument(split instanceof ChainSplit);
-            return fallbackRead.createReader(split);
+            if (split instanceof ChainSplit || split instanceof DataSplit) {
+                // Both ChainSplit (full load) and DataSplit (incremental streaming)
+                // are handled by fallbackRead (delta table's MergeFileSplitRead):
+                // - ChainSplit → createChainReader() for merge-engine dedup
+                // - DataSplit(isStreaming=true) → createNoMergeReader() for changelog passthrough
+                return fallbackRead.createReader(split);
+            }
+            throw new IllegalArgumentException(
+                    "Unsupported split type for chain table read: " + split.getClass().getName());
         }
     }
 }

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

@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.ChangelogProducer;
+import org.apache.paimon.CoreOptions.StartupMode;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.ReadBuilderImpl;
+import org.apache.paimon.table.source.StreamDataTableScan;
+
+import java.util.Map;
+
+/**
+ * Chain-table-aware wrapper for {@link FileStoreTable}. Wraps the {@link
+ * FallbackReadFileStoreTable} that already handles batch reads via {@link ChainGroupReadTable}, and
+ * additionally overrides {@link #newStreamScan()} to return a chain-aware {@link
+ * ChainTableStreamScan} that performs a full chain-merge load followed by incremental delta-only
+ * streaming.
+ *
+ * <p>This class exists because {@link FallbackReadFileStoreTable} is a general-purpose fallback
+ * mechanism (used by {@code scan.fallback-branch}, {@code scan.primary-branch}, and chain tables).
+ * Chain-table-specific streaming logic is isolated here to avoid polluting the generic fallback.
+ */
+public class ChainTableFileStoreTable extends DelegatedFileStoreTable {
+
+    public ChainTableFileStoreTable(FileStoreTable wrapped) {
+        super(wrapped);
+    }
+
+    @Override
+    public StreamDataTableScan newStreamScan() {
+        CoreOptions coreOptions = wrapped.coreOptions();
+
+        // When the user specifies an explicit starting position (e.g., scan.snapshot-id,
+        // scan.timestamp-millis, scan.mode=latest, consumer-id with existing progress),
+        // delegate to the standard streaming scan. The chain-table-aware scan only supports
+        // the default behavior: full chain-merge load then incremental delta streaming.
+        StartupMode effectiveMode = coreOptions.startupMode();
+        boolean hasConsumerProgress =
+                coreOptions.consumerId() != null && !coreOptions.consumerIgnoreProgress();
+        if (effectiveMode != StartupMode.LATEST_FULL || hasConsumerProgress) {
+            return wrapped.newStreamScan();
+        }
+
+        // Navigate through the wrapper hierarchy:
+        //   wrapped = FallbackReadFileStoreTable
+        //     other() = ChainGroupReadTable
+        FileStoreTable other = ((FallbackReadFileStoreTable) wrapped).other();
+        ChainGroupReadTable chainGroupReadTable = (ChainGroupReadTable) other;
+
+        // Chain table streaming read requires a changelog producer on the delta branch.
+        // Without changelog, the incremental phase cannot correctly propagate changes.
+        FileStoreTable deltaTable = chainGroupReadTable.other();
+        ChangelogProducer changelogProducer = deltaTable.coreOptions().changelogProducer();
+        if (changelogProducer == ChangelogProducer.NONE) {
+            throw new IllegalArgumentException(
+                    "Chain table streaming read requires a changelog producer "
+                            + "(e.g., 'changelog-producer' = 'input') on the delta branch. "
+                            + "Currently the delta branch has 'changelog-producer' = 'none'.");
+        }
+
+        TableSchema schema = ((AbstractFileStoreTable) chainGroupReadTable.wrapped()).tableSchema;
+        return new ChainTableStreamScan(chainGroupReadTable, schema);
+    }
+
+    @Override
+    public ReadBuilder newReadBuilder() {
+        return new ReadBuilderImpl(this) {
+            @Override
+            public StreamDataTableScan newStreamScan() {
+                return ChainTableFileStoreTable.this.newStreamScan();
+            }
+        };
+    }
+
+    @Override
+    public FileStoreTable copy(Map<String, String> dynamicOptions) {
+        return new ChainTableFileStoreTable(wrapped.copy(dynamicOptions));
+    }
+
+    @Override
+    public FileStoreTable copy(TableSchema newTableSchema) {
+        return new ChainTableFileStoreTable(wrapped.copy(newTableSchema));
+    }
+
+    @Override
+    public FileStoreTable copyWithoutTimeTravel(Map<String, String> dynamicOptions) {
+        return new ChainTableFileStoreTable(wrapped.copyWithoutTimeTravel(dynamicOptions));
+    }
+
+    @Override
+    public FileStoreTable copyWithLatestSchema() {
+        return new ChainTableFileStoreTable(wrapped.copyWithLatestSchema());
+    }
+
+    @Override
+    public FileStoreTable switchToBranch(String branchName) {
+        return new ChainTableFileStoreTable(wrapped.switchToBranch(branchName));
+    }
+}