This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4badbf69df [core] Split chain table batch scans by key range for read
parallelism. (#8831)
4badbf69df is described below
commit 4badbf69df996a18c46bcc010868aa58731e970a
Author: Wenchao Wu <[email protected]>
AuthorDate: Sun Jul 26 15:52:59 2026 +0800
[core] Split chain table batch scans by key range for read parallelism.
(#8831)
---
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 16 +++
.../apache/paimon/table/ChainGroupReadTable.java | 27 +++-
.../org/apache/paimon/utils/ChainTableUtils.java | 133 ++++++++++++++++--
.../apache/paimon/utils/ChainTableUtilsTest.java | 150 +++++++++++++++++++++
5 files changed, 323 insertions(+), 9 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index bd739d57a4..630b6eb200 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1386,6 +1386,12 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<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>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index a57810ae45..6a06832770 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -314,6 +314,18 @@ public class CoreOptions implements Serializable {
+ "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";
@@ -4296,6 +4308,10 @@ public class CoreOptions implements Serializable {
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;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
index c1e67e8a95..cf3c32be27 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
@@ -49,6 +49,7 @@ import java.io.IOException;
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 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
snapshotSubSplits,
deltaSubSplits,
options.scanFallbackSnapshotBranch(),
- options.scanFallbackDeltaBranch()));
+ options.scanFallbackDeltaBranch(),
+ keyComparator,
+ targetSplitSize,
+ openFileCost));
}
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
index 498cc4dacc..35659f06d8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
@@ -21,8 +21,11 @@ package org.apache.paimon.utils;
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.ChainSplit;
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 class ChainTableUtils {
* 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 class ChainTableUtils {
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 class ChainTableUtils {
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 class ChainTableUtils {
? 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();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
index 971407b928..50e74d4d4e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
@@ -24,9 +24,15 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.stats.StatsTestUtils;
+import org.apache.paimon.table.source.ChainSplit;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -37,9 +43,12 @@ import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -767,4 +776,145 @@ public class ChainTableUtilsTest {
assertThat(getString(matched2, 0)).isEqualTo("20250809");
assertThat(getString(matched2, 1)).isEqualTo("02");
}
+
+ @Test
+ public void testBuildChainSplitsWithKeyRangeSplitting() {
+ // Single bucket. Snapshot files S1=[a,b], S2=[m,n]; delta files
D1=[a,b] (overlaps S1),
+ // D2=[m,n] (overlaps S2). The two sections are key-disjoint, so they
form two splits;
+ // within each section the overlapping snapshot and delta files must
stay together so that
+ // all versions of a key are merged in the same split.
+ BinaryRow partition = row(Lists.newArrayList("p0"));
+ String snapBranch = "snap";
+ String deltaBranch = "delta";
+
+ DataFileMeta s1 = makeFile("S1", "a", "b", 100);
+ DataFileMeta s2 = makeFile("S2", "m", "n", 100);
+ DataFileMeta d1 = makeFile("D1", "a", "b", 100);
+ DataFileMeta d2 = makeFile("D2", "m", "n", 100);
+
+ DataSplit snapshotSplit = dataSplit(partition, 0,
Lists.newArrayList(s1, s2));
+ DataSplit deltaSplit = dataSplit(partition, 0, Lists.newArrayList(d1,
d2));
+
+ // Small targetSplitSize so each section forms its own split.
+ List<ChainSplit> splits =
+ ChainTableUtils.buildChainSplits(
+ partition,
+ Collections.singletonList(snapshotSplit),
+ Collections.singletonList(deltaSplit),
+ snapBranch,
+ deltaBranch,
+ KEY_COMPARATOR,
+ 1L,
+ 1L);
+
+ assertThat(splits).hasSize(2);
+
+ List<Set<String>> groups =
+
splits.stream().map(ChainTableUtilsTest::fileNames).collect(Collectors.toList());
+ // Overlapping files are kept together: {S1,D1} and {S2,D2}, never
mixed.
+ assertThat(groups)
+ .containsExactlyInAnyOrder(
+ new HashSet<>(Arrays.asList("S1", "D1")),
+ new HashSet<>(Arrays.asList("S2", "D2")));
+
+ // Branch tagging is preserved per file across the resulting splits.
+ ChainSplit s1Split =
+ splits.stream().filter(s ->
fileNames(s).contains("S1")).findFirst().get();
+
assertThat(s1Split.fileBranchMapping().get("S1")).isEqualTo(snapBranch);
+
assertThat(s1Split.fileBranchMapping().get("D1")).isEqualTo(deltaBranch);
+ ChainSplit s2Split =
+ splits.stream().filter(s ->
fileNames(s).contains("S2")).findFirst().get();
+
assertThat(s2Split.fileBranchMapping().get("S2")).isEqualTo(snapBranch);
+
assertThat(s2Split.fileBranchMapping().get("D2")).isEqualTo(deltaBranch);
+ }
+
+ @Test
+ public void
testBuildChainSplitsKeyRangeSplittingLargeTargetKeepsOneSplit() {
+ // With a large targetSplitSize all key-disjoint sections are packed
into a single split.
+ BinaryRow partition = row(Lists.newArrayList("p0"));
+ DataFileMeta s1 = makeFile("S1", "a", "b", 100);
+ DataFileMeta s2 = makeFile("S2", "m", "n", 100);
+ DataFileMeta d1 = makeFile("D1", "a", "b", 100);
+ DataFileMeta d2 = makeFile("D2", "m", "n", 100);
+
+ DataSplit snapshotSplit = dataSplit(partition, 0,
Lists.newArrayList(s1, s2));
+ DataSplit deltaSplit = dataSplit(partition, 0, Lists.newArrayList(d1,
d2));
+
+ List<ChainSplit> splits =
+ ChainTableUtils.buildChainSplits(
+ partition,
+ Collections.singletonList(snapshotSplit),
+ Collections.singletonList(deltaSplit),
+ "snap",
+ "delta",
+ KEY_COMPARATOR,
+ Long.MAX_VALUE,
+ 1L);
+
+ assertThat(splits).hasSize(1);
+ assertThat(fileNames(splits.get(0))).containsExactlyInAnyOrder("S1",
"S2", "D1", "D2");
+ }
+
+ @Test
+ public void testBuildChainSplitsWithoutKeyRangeSplitting() {
+ // A null keyComparator keeps the original one-split-per-bucket
behavior (used by
+ // streaming).
+ BinaryRow partition = row(Lists.newArrayList("p0"));
+ DataFileMeta s1 = makeFile("S1", "a", "b", 100);
+ DataFileMeta d1 = makeFile("D1", "a", "b", 100);
+ DataSplit snapshotSplit = dataSplit(partition, 0,
Lists.newArrayList(s1));
+ DataSplit deltaSplit = dataSplit(partition, 0, Lists.newArrayList(d1));
+
+ List<ChainSplit> splits =
+ ChainTableUtils.buildChainSplits(
+ partition,
+ Collections.singletonList(snapshotSplit),
+ Collections.singletonList(deltaSplit),
+ "snap",
+ "delta",
+ null,
+ 0L,
+ 0L);
+
+ assertThat(splits).hasSize(1);
+ assertThat(fileNames(splits.get(0))).containsExactlyInAnyOrder("S1",
"D1");
+ }
+
+ private static DataFileMeta makeFile(String name, String min, String max,
long size) {
+ return DataFileMeta.create(
+ name,
+ size,
+ 10L,
+ row(Lists.newArrayList(min)),
+ row(Lists.newArrayList(max)),
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ 0L,
+ 9L,
+ 0L,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(1000L),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static DataSplit dataSplit(BinaryRow partition, int bucket,
List<DataFileMeta> files) {
+ return DataSplit.builder()
+ .withPartition(partition)
+ .withBucket(bucket)
+ .withBucketPath("bucket_" + bucket)
+ .withTotalBuckets(1)
+ .withDataFiles(files)
+ .build();
+ }
+
+ private static Set<String> fileNames(ChainSplit split) {
+ return
split.dataFiles().stream().map(DataFileMeta::fileName).collect(Collectors.toSet());
+ }
}
