[core] Split chain table batch scans by key range for read parallelism.

docs/generated/core_configuration.html

@@ -1361,6 +1361,12 @@
             <td>Long</td>
             <td>After configuring this time, only the data files created after this time will be read. It is independent of snapshots, but it is imprecise filtering (depending on whether or not compaction occurs).</td>
         </tr>
+        <tr>
+            <td><h5>chain-table.split.key-range-enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>If true, a batch chain-table scan splits each bucket's snapshot and delta files into multiple splits by key range to improve read parallelism. Files with intersecting key ranges always stay in the same split so that all versions of a key across the snapshot and delta branches are merged together. Set to false to fall back to one split per bucket.</td>
+        </tr>
         <tr>
             <td><h5>scan.ignore-corrupt-files</h5></td>
             <td style="word-wrap: break-word;">false</td>

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

@@ -311,6 +311,18 @@ public InlineElement getDescription() {
                                     + "splits, which is lightweight but may not reflect cross-branch "
                                     + "deletes.");
 
+    public static final ConfigOption<Boolean> CHAIN_TABLE_KEY_RANGE_SPLIT_ENABLED =
+            key("chain-table.split.key-range-enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "If true, a batch chain-table scan splits each bucket's snapshot and "
+                                    + "delta files into multiple splits by key range to improve read "
+                                    + "parallelism. Files with intersecting key ranges always stay in "
+                                    + "the same split so that all versions of a key across the "
+                                    + "snapshot and delta branches are merged together. Set to false "
+                                    + "to fall back to one split per bucket.");
+
     public static final String FILE_FORMAT_ORC = "orc";
     public static final String FILE_FORMAT_AVRO = "avro";
     public static final String FILE_FORMAT_PARQUET = "parquet";
@@ -4228,6 +4240,10 @@ public boolean chainTableStreamingMergeSnapshot() {
         return options.get(CHAIN_TABLE_STREAMING_MERGE_SNAPSHOT);
     }
 
+    public boolean chainTableKeyRangeSplitEnabled() {
+        return options.get(CHAIN_TABLE_KEY_RANGE_SPLIT_ENABLED);
+    }
+
     public boolean formatTableImplementationIsPaimon() {
         return options.get(FORMAT_TABLE_IMPLEMENTATION) == FormatTableImplementation.PAIMON;
     }

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

@@ -49,6 +49,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
@@ -119,6 +120,16 @@ private DataTableScan newDeltaScan(Function<FileStoreTable, DataTableScan> scanC
         return scanCreator.apply(other());
     }
 
+    /**
+     * Returns the primary-key comparator of the snapshot branch. It is used by batch scans to split
+     * each bucket's snapshot and delta files into key-range splits. Both branches share the same
+     * primary-key schema, so the snapshot branch's comparator correctly orders keys from either
+     * branch.
+     */
+    Comparator<InternalRow> chainKeyComparator() {
+        return ((PrimaryKeyFileStoreTable) wrapped).store().newKeyComparator();
+    }
+
     @Override
     public FileStoreTable copy(Map<String, String> dynamicOptions) {
         Map<String, String> wrappedOptions =
@@ -336,6 +347,17 @@ public Plan plan() {
             PredicateBuilder builder = new PredicateBuilder(tableSchema.logicalPartitionType());
             Set<BinaryRow> snapshotPartitions = preloadTargetSnapshotSplits(splits);
 
+            // Key-range splitting parameters are loop-invariant; compute them once. When key-range
+            // splitting is enabled, each bucket's snapshot and delta files are split into multiple
+            // splits to improve read parallelism (files with intersecting key ranges stay
+            // together).
+            Comparator<InternalRow> keyComparator =
+                    options.chainTableKeyRangeSplitEnabled()
+                            ? chainGroupReadTable.chainKeyComparator()
+                            : null;
+            long targetSplitSize = options.splitTargetSize();
+            long openFileCost = options.splitOpenFileCost();
+
             DataTableScan deltaPartitionScan =
                     newChainPartitionListingScan(false, getFallbackPartitionPredicate());
             List<BinaryRow> deltaPartitions =
@@ -449,7 +471,10 @@ public Plan plan() {
                                         snapshotSubSplits,
                                         deltaSubSplits,
                                         options.scanFallbackSnapshotBranch(),
-                                        options.scanFallbackDeltaBranch()));
+                                        options.scanFallbackDeltaBranch(),
+                                        keyComparator,
+                                        targetSplitSize,
+                                        openFileCost));
                     }
                 }
             }

paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java

@@ -21,8 +21,11 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.codegen.RecordComparator;
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.serializer.InternalRowSerializer;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.mergetree.SortedRun;
+import org.apache.paimon.mergetree.compact.IntervalPartition;
 import org.apache.paimon.partition.PartitionTimeExtractor;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateBuilder;
@@ -33,15 +36,20 @@
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.types.RowType;
 
+import javax.annotation.Nullable;
+
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -377,6 +385,9 @@ public static Predicate createGroupChainPredicate(
      * originate from the snapshot splits are tagged with {@code snapshotBranch}; all other files
      * are tagged with {@code deltaBranch}.
      *
+     * <p>This overload produces one {@link ChainSplit} per bucket (no key-range splitting). It is
+     * used by streaming read, where splitting a bucket by key range is not desired.
+     *
      * @param logicalPartition the logical partition for the resulting ChainSplits
      * @param snapshotSplits splits from the snapshot branch
      * @param deltaSplits splits from the delta branch
@@ -390,6 +401,49 @@ public static List<ChainSplit> buildChainSplits(
             List<DataSplit> deltaSplits,
             String snapshotBranch,
             String deltaBranch) {
+        return buildChainSplits(
+                logicalPartition,
+                snapshotSplits,
+                deltaSplits,
+                snapshotBranch,
+                deltaBranch,
+                null,
+                0L,
+                0L);
+    }
+
+    /**
+     * Builds {@link ChainSplit}s from the given snapshot and delta splits, optionally splitting
+     * each bucket's files into multiple splits by key range to improve batch read parallelism.
+     *
+     * <p>When {@code keyComparator} is non-null, each bucket's snapshot and delta files are split
+     * via interval partition: files with intersecting key ranges always go to the same split, so
+     * that all versions of a key across the snapshot and delta branches are merged together within
+     * one split. Key-disjoint sections are then packed into splits up to {@code targetSplitSize}.
+     * This is the same invariant the ordinary primary-key table relies on (see {@link
+     * org.apache.paimon.table.source.MergeTreeSplitGenerator}), except the chain path always goes
+     * through the interval partition because snapshot and delta files from different branches may
+     * have intersecting key ranges.
+     *
+     * <p>When {@code keyComparator} is null, each bucket produces a single {@link ChainSplit} (the
+     * original behavior, used by streaming).
+     *
+     * @param keyComparator primary-key comparator used to order files by min/max key; null disables
+     *     key-range splitting
+     * @param targetSplitSize target size (in bytes) of each split; ignored when {@code
+     *     keyComparator} is null
+     * @param openFileCost per-file open cost accounted for in packing; ignored when {@code
+     *     keyComparator} is null
+     */
+    public static List<ChainSplit> buildChainSplits(
+            BinaryRow logicalPartition,
+            List<DataSplit> snapshotSplits,
+            List<DataSplit> deltaSplits,
+            String snapshotBranch,
+            String deltaBranch,
+            @Nullable Comparator<InternalRow> keyComparator,
+            long targetSplitSize,
+            long openFileCost) {
         Set<String> snapshotFileNames =
                 snapshotSplits.stream()
                         .flatMap(s -> s.dataFiles().stream().map(DataFileMeta::fileName))
@@ -408,6 +462,7 @@ public static List<ChainSplit> buildChainSplits(
         for (Map.Entry<Integer, List<DataSplit>> entry : bucketSplits.entrySet()) {
             Map<String, String> fileBranchMapping = new HashMap<>();
             Map<String, String> fileBucketPathMapping = new HashMap<>();
+            List<DataFileMeta> bucketFiles = new ArrayList<>();
             for (DataSplit ds : entry.getValue()) {
                 for (DataFileMeta file : ds.dataFiles()) {
                     fileBucketPathMapping.put(file.fileName(), ds.bucketPath());
@@ -416,20 +471,82 @@ public static List<ChainSplit> buildChainSplits(
                                     ? snapshotBranch
                                     : deltaBranch;
                     fileBranchMapping.put(file.fileName(), branch);
+                    bucketFiles.add(file);
                 }
             }
-            result.add(
-                    new ChainSplit(
-                            logicalPartition,
-                            entry.getValue().stream()
-                                    .flatMap(ds -> ds.dataFiles().stream())
-                                    .collect(Collectors.toList()),
-                            fileBranchMapping,
-                            fileBucketPathMapping));
+
+            List<List<DataFileMeta>> fileGroups;
+            if (keyComparator == null) {
+                fileGroups = Collections.singletonList(bucketFiles);
+            } else {
+                fileGroups =
+                        splitBucketByKeyRange(
+                                bucketFiles, keyComparator, targetSplitSize, openFileCost);
+            }
+
+            for (List<DataFileMeta> groupFiles : fileGroups) {
+                result.add(
+                        new ChainSplit(
+                                logicalPartition,
+                                groupFiles,
+                                subMapping(fileBranchMapping, groupFiles),
+                                subMapping(fileBucketPathMapping, groupFiles)));
+            }
         }
         return result;
     }
 
+    /**
+     * Splits a bucket's files (snapshot + delta combined) into key-range groups. The interval
+     * partition first sorts files by (minKey, maxKey) and groups key-overlapping files into the
+     * same section, guaranteeing all versions of a key stay in one split. Sections are key-disjoint
+     * and key-ordered, so packing consecutive sections up to {@code targetSplitSize} preserves that
+     * invariant.
+     */
+    private static List<List<DataFileMeta>> splitBucketByKeyRange(
+            List<DataFileMeta> files,
+            Comparator<InternalRow> keyComparator,
+            long targetSplitSize,
+            long openFileCost) {
+        List<List<DataFileMeta>> sections = new ArrayList<>();
+        for (List<SortedRun> section : new IntervalPartition(files, keyComparator).partition()) {
+            List<DataFileMeta> sectionFiles = new ArrayList<>();
+            for (SortedRun run : section) {
+                sectionFiles.addAll(run.files());
+            }
+            sections.add(sectionFiles);
+        }
+
+        Function<List<DataFileMeta>, Long> weightFunc =
+                section -> Math.max(totalSize(section), openFileCost);
+        return BinPacking.packForOrdered(sections, weightFunc, targetSplitSize).stream()
+                .map(ChainTableUtils::flatFiles)
+                .collect(Collectors.toList());
+    }
+
+    private static long totalSize(List<DataFileMeta> files) {
+        long size = 0L;
+        for (DataFileMeta file : files) {
+            size += file.fileSize();
+        }
+        return size;
+    }
+
+    private static List<DataFileMeta> flatFiles(List<List<DataFileMeta>> packedSections) {
+        List<DataFileMeta> files = new ArrayList<>();
+        packedSections.forEach(files::addAll);
+        return files;
+    }
+
+    private static Map<String, String> subMapping(
+            Map<String, String> mapping, List<DataFileMeta> files) {
+        Map<String, String> sub = new HashMap<>();
+        for (DataFileMeta file : files) {
+            sub.put(file.fileName(), mapping.get(file.fileName()));
+        }
+        return sub;
+    }
+
     private static Integer addToBucketMap(
             DataSplit ds, Map<Integer, List<DataSplit>> bucketSplits, Integer bucketInAll) {
         Integer totalBuckets = ds.totalBuckets();