[flink] Support stream read Chain Table

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.
 - Deletion vector is not supported for chain table.
 
 After creating a chain table, you can read and write data in the following ways.
@@ -216,6 +215,55 @@ you will get the following result:
 +---+----+-----+
 ```
 
+## Streaming Read
+
+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.
+2. **Incremental phase**: Continuously reads new commits from the delta branch as they arrive.
+
+### 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 streaming read relies on this pattern to produce correct results. After the full load phase,
+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` — an
+  `UnsupportedOperationException` is thrown. To use standard streaming read without chain table
+  logic, read from a specific branch table (e.g., `t$branch_delta`) instead of the main table.
+- Partition filters are not supported in chain table streaming reads. Specifying a partition
+  filter — either via a `WHERE` clause on partition columns or the `scan.partitions` table
+  option — throws an `UnsupportedOperationException`. This is because the chain table streaming
+  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.
+
 ## 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/table/ChainGroupReadTable.java

@@ -535,8 +535,11 @@ 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) {
+                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,219 @@
+/*
+ * 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.StartupMode;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.source.ChainSplit;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.RowType;
+
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Chain-table-aware extension of {@link FallbackReadFileStoreTable}. Inherits the batch read
+ * behavior (partition-level fallback between the current branch and {@link ChainGroupReadTable}),
+ * and additionally overrides {@link #newStreamScan()} to return a chain-aware {@link
+ * ChainTableStreamScan} that performs a partition-level full load followed by incremental
+ * delta-only streaming.
+ */
+public class ChainTableFileStoreTable extends FallbackReadFileStoreTable {
+
+    public ChainTableFileStoreTable(FileStoreTable wrapped, FileStoreTable other) {
+        super(wrapped, other, true);
+    }
+
+    @Override
+    public StreamDataTableScan newStreamScan() {
+        CoreOptions coreOptions = wrapped.coreOptions();
+
+        StartupMode effectiveMode = coreOptions.startupMode();
+        boolean hasConsumer = coreOptions.consumerId() != null;
+        if (effectiveMode != StartupMode.LATEST_FULL || hasConsumer) {
+            String reason = describeUnsupportedMode(coreOptions, effectiveMode, hasConsumer);
+            throw new UnsupportedOperationException(
+                    "Chain table streaming read does not support startup mode '"
+                            + reason
+                            + "'. "
+                            + "Chain table streaming only supports the default 'latest-full' mode, which first "
+                            + "produces a partition-level full result and then continuously reads incremental "
+                            + "data from the delta branch.\n"
+                            + "Suggestions:\n"
+                            + "  - To use chain table streaming: remove the explicit scan mode/position settings "
+                            + "so that the default 'latest-full' mode is used.\n"
+                            + "  - To use standard streaming read without chain table logic: read from a "
+                            + "specific branch table (e.g., 't$branch_delta') instead of the main table.");
+        }
+
+        // Inherited other() returns the ChainGroupReadTable directly.
+        ChainGroupReadTable chainGroupReadTable = (ChainGroupReadTable) other();
+
+        return new ChainTableStreamScan(chainGroupReadTable);
+    }
+
+    private static String describeUnsupportedMode(
+            CoreOptions coreOptions, StartupMode effectiveMode, boolean hasConsumer) {
+        if (hasConsumer) {
+            return "consumer mode (consumer-id='" + coreOptions.consumerId() + "')";
+        }
+        switch (effectiveMode) {
+            case LATEST:
+                return "scan.mode=latest";
+            case FROM_SNAPSHOT:
+                if (coreOptions.scanSnapshotId() != null) {
+                    return "scan.snapshot-id=" + coreOptions.scanSnapshotId();
+                }
+                if (coreOptions.scanTagName() != null) {
+                    return "scan.tag-name=" + coreOptions.scanTagName();
+                }
+                if (coreOptions.scanWatermark() != null) {
+                    return "scan.watermark=" + coreOptions.scanWatermark();
+                }
+                return "from-snapshot";
+            case FROM_TIMESTAMP:
+                if (coreOptions.scanTimestampMills() != null) {
+                    return "scan.timestamp-millis=" + coreOptions.scanTimestampMills();
+                }
+                if (coreOptions.scanTimestamp() != null) {
+                    return "scan.timestamp=" + coreOptions.scanTimestamp();
+                }
+                return "from-timestamp";
+            default:
+                return effectiveMode.name().toLowerCase().replace('_', '-');
+        }
+    }
+
+    @Override
+    public InnerTableRead newRead() {
+        return new ChainTableRead();
+    }
+
+    @Override
+    public FileStoreTable copy(Map<String, String> dynamicOptions) {
+        return new ChainTableFileStoreTable(
+                wrapped.copy(dynamicOptions), other().copy(rewriteOtherOptions(dynamicOptions)));
+    }
+
+    @Override
+    public FileStoreTable copy(TableSchema newTableSchema) {
+        return new ChainTableFileStoreTable(
+                wrapped.copy(newTableSchema),
+                other().copy(newTableSchema.copy(rewriteOtherOptions(newTableSchema.options()))));
+    }
+
+    @Override
+    public FileStoreTable copyWithoutTimeTravel(Map<String, String> dynamicOptions) {
+        return new ChainTableFileStoreTable(
+                wrapped.copyWithoutTimeTravel(dynamicOptions),
+                other().copyWithoutTimeTravel(rewriteOtherOptions(dynamicOptions)));
+    }
+
+    @Override
+    public FileStoreTable copyWithLatestSchema() {
+        return new ChainTableFileStoreTable(
+                wrapped.copyWithLatestSchema(), other().copyWithLatestSchema());
+    }
+
+    @Override
+    public FileStoreTable switchToBranch(String branchName) {
+        return new ChainTableFileStoreTable(switchWrappedToBranch(branchName), other());
+    }
+
+    /**
+     * Chain-aware read implementation that pairs with {@link ChainTableStreamScan}. Routes splits
+     * based on type:
+     *
+     * <ul>
+     *   <li><b>ChainSplit / DataSplit:</b> Streaming read splits. Routed to {@link
+     *       ChainGroupReadTable}'s read which uses {@link
+     *       org.apache.paimon.io.ChainKeyValueFileReaderFactory} (both branch schemas for
+     *       ChainSplit) or delta branch read (DataSplit with correct schema).
+     *   <li><b>FallbackSplit:</b> Batch read fallback splits. Routed to inherited {@link
+     *       FallbackReadFileStoreTable} read for partition-level fallback logic.
+     * </ul>
+     */
+    private class ChainTableRead implements InnerTableRead {
+
+        private final InnerTableRead chainGroupRead;
+        private final InnerTableRead fallbackRead;
+
+        private ChainTableRead() {
+            this.chainGroupRead = other().newRead();
+            this.fallbackRead = ChainTableFileStoreTable.super.newRead();
+        }
+
+        @Override
+        public InnerTableRead withFilter(Predicate predicate) {
+            chainGroupRead.withFilter(predicate);
+            fallbackRead.withFilter(predicate);
+            return this;
+        }
+
+        @Override
+        public InnerTableRead withReadType(RowType readType) {
+            chainGroupRead.withReadType(readType);
+            fallbackRead.withReadType(readType);
+            return this;
+        }
+
+        @Override
+        public InnerTableRead forceKeepDelete() {
+            chainGroupRead.forceKeepDelete();
+            fallbackRead.forceKeepDelete();
+            return this;
+        }
+
+        @Override
+        public TableRead executeFilter() {
+            chainGroupRead.executeFilter();
+            fallbackRead.executeFilter();
+            return this;
+        }
+
+        @Override
+        public TableRead withIOManager(IOManager ioManager) {
+            chainGroupRead.withIOManager(ioManager);
+            fallbackRead.withIOManager(ioManager);
+            return this;
+        }
+
+        @Override
+        public RecordReader<InternalRow> createReader(Split split) throws IOException {
+            if (split instanceof FallbackSplit) {
+                // FallbackSplit (including FallbackDataSplit): use inherited fallback read logic
+                return fallbackRead.createReader(split);
+            }
+            if (split instanceof ChainSplit || split instanceof DataSplit) {
+                return chainGroupRead.createReader(split);
+            }
+            // Other split types: use inherited fallback read logic
+            return fallbackRead.createReader(split);
+        }
+    }
+}