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 dd1b91d57d [core] Avoid repeated sorted-index source lookups (#9187)
dd1b91d57d is described below
commit dd1b91d57d9ef5d519c956d511f826e1dacd520d
Author: QuakeWang <[email protected]>
AuthorDate: Thu Aug 13 22:01:02 2026 +0800
[core] Avoid repeated sorted-index source lookups (#9187)
---
.../table/source/PrimaryKeySortedIndexScan.java | 55 ++++++++++++++--------
.../source/PrimaryKeySortedIndexScanTest.java | 27 +++++++----
2 files changed, 53 insertions(+), 29 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
index 7bc04f5d6f..cb03736b7b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java
@@ -54,6 +54,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
@@ -274,8 +275,7 @@ public final class PrimaryKeySortedIndexScan {
totalRowCount));
sharedReaders.put(indexGroup, reader);
}
- return Collections.singletonList(
- fileLocalReader(file, group.get(),
reader));
+ return
Collections.singletonList(fileLocalReader(file, reader));
});
Optional<GlobalIndexResult> result;
try {
@@ -300,23 +300,20 @@ public final class PrimaryKeySortedIndexScan {
}
private static GlobalIndexReader fileLocalReader(
- FilePlan file, PkSortedIndexGroup group, SharedGlobalIndexReader
reader) {
- List<PrimaryKeyIndexSourceFile> sourceFiles = group.sourceFiles();
- PrimaryKeyIndexSourceFile target =
- new PrimaryKeyIndexSourceFile(
- file.dataFile().fileName(),
file.dataFile().rowCount());
- int sourceIndex = -1;
- for (int i = 0; i < sourceFiles.size(); i++) {
- if (sourceFiles.get(i).equals(target)) {
- sourceIndex = i;
- break;
- }
- }
+ FilePlan file, SharedGlobalIndexReader reader) {
+ DataFileMeta dataFile = file.dataFile();
+ SourceLocation sourceLocation =
reader.sourceLocations.get(dataFile.fileName());
checkArgument(
- sourceIndex >= 0,
+ sourceLocation != null,
"Data file %s is not covered by its sorted-index source
group.",
- file.dataFile().fileName());
- return new FileLocalGlobalIndexReader(reader, sourceIndex);
+ dataFile.fileName());
+ checkArgument(
+ dataFile.rowCount() == sourceLocation.rowCount,
+ "Data file %s row count %s does not match sorted-index source
row count %s.",
+ dataFile.fileName(),
+ dataFile.rowCount(),
+ sourceLocation.rowCount);
+ return new FileLocalGlobalIndexReader(reader,
sourceLocation.sourceIndex);
}
private static long totalRowCount(List<PrimaryKeyIndexSourceFile>
sourceFiles) {
@@ -333,6 +330,17 @@ public final class PrimaryKeySortedIndexScan {
}
}
+ private static final class SourceLocation {
+
+ private final long rowCount;
+ private final int sourceIndex;
+
+ private SourceLocation(long rowCount, int sourceIndex) {
+ this.rowCount = rowCount;
+ this.sourceIndex = sourceIndex;
+ }
+ }
+
/** Shares one source-group reader and its group-global query results
across source files. */
private static final class SharedGlobalIndexReader implements
GlobalIndexReader {
@@ -342,6 +350,7 @@ public final class PrimaryKeySortedIndexScan {
CompletableFuture<Optional<GlobalIndexResult>>,
CompletableFuture<List<Optional<GlobalIndexResult>>>>
localizedResults;
+ private final Map<String, SourceLocation> sourceLocations;
private final long[] sourceOffsets;
private GlobalIndexReader reader;
@@ -353,10 +362,18 @@ public final class PrimaryKeySortedIndexScan {
this.readerFactory = readerFactory;
this.results = new ConcurrentHashMap<>();
this.localizedResults = new ConcurrentHashMap<>();
+ this.sourceLocations = new HashMap<>();
this.sourceOffsets = new long[sourceFiles.size() + 1];
for (int i = 0; i < sourceFiles.size(); i++) {
- sourceOffsets[i + 1] =
- Math.addExact(sourceOffsets[i],
sourceFiles.get(i).rowCount());
+ PrimaryKeyIndexSourceFile sourceFile = sourceFiles.get(i);
+ checkArgument(
+ sourceLocations.put(
+ sourceFile.fileName(),
+ new
SourceLocation(sourceFile.rowCount(), i))
+ == null,
+ "Duplicate sorted-index source file %s.",
+ sourceFile.fileName());
+ sourceOffsets[i + 1] = Math.addExact(sourceOffsets[i],
sourceFile.rowCount());
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
index c704589af7..b800fb1c61 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScanTest.java
@@ -205,7 +205,8 @@ class PrimaryKeySortedIndexScanTest {
void testReadMergedSourceGroupInFileLocalPositions() throws IOException {
DataFileMeta first = dataFile("data-1", 2);
DataFileMeta second = dataFile("data-2", 3);
- DataSplit split = dataSplit(11, 0, true, second, first);
+ DataFileMeta third = dataFile("data-3", 4);
+ DataSplit split = dataSplit(11, 0, true, third, first, second);
PrimaryKeyIndexDefinition definition =
definition(
7,
@@ -216,10 +217,11 @@ class PrimaryKeySortedIndexScanTest {
"btree-merged",
Arrays.asList(
new PrimaryKeyIndexSourceFile("data-1", 2),
- new PrimaryKeyIndexSourceFile("data-2", 3)),
+ new PrimaryKeyIndexSourceFile("data-2", 3),
+ new PrimaryKeyIndexSourceFile("data-3", 4)),
"btree",
7,
- 5);
+ 9);
PrimaryKeySortedIndexScan.Plan plan =
PrimaryKeySortedIndexScan.plan(
11,
@@ -232,8 +234,10 @@ class PrimaryKeySortedIndexScanTest {
AtomicInteger queries = new AtomicInteger();
CountingRoaringNavigableMap64 groupPositions = new
CountingRoaringNavigableMap64();
groupPositions.add(1);
- groupPositions.add(3);
+ groupPositions.add(2);
groupPositions.add(4);
+ groupPositions.add(6);
+ groupPositions.add(7);
GlobalIndexReader reader = mock(GlobalIndexReader.class);
when(reader.visitEqual(any(), eq(42)))
.thenAnswer(
@@ -251,23 +255,26 @@ class PrimaryKeySortedIndexScanTest {
(ignoredFile, ignoredDefinition, payloads,
totalRowCount) -> {
readersCreated.incrementAndGet();
assertThat(payloads).containsExactly(mergedPayload);
- assertThat(totalRowCount).isEqualTo(5);
+ assertThat(totalRowCount).isEqualTo(9);
return reader;
});
PrimaryKeySortedIndexResult result = new
PrimaryKeySortedIndexResult(evaluated);
assertThat(readersCreated).hasValue(1);
assertThat(queries).hasValue(1);
- assertThat(groupPositions.iteratedPositions()).isEqualTo(3);
+ assertThat(groupPositions.iteratedPositions()).isEqualTo(5);
verify(reader, times(1)).close();
- assertThat(result.splits()).hasSize(2);
+ assertThat(result.splits()).hasSize(3);
assertThat(result.splits()).allMatch(IndexedSplit.class::isInstance);
- IndexedSplit secondSplit = (IndexedSplit) result.splits().get(0);
-
assertThat(secondSplit.dataSplit().dataFiles()).containsExactly(second);
- assertThat(secondSplit.rowRanges()).containsExactly(new Range(1, 2));
+ IndexedSplit thirdSplit = (IndexedSplit) result.splits().get(0);
+ assertThat(thirdSplit.dataSplit().dataFiles()).containsExactly(third);
+ assertThat(thirdSplit.rowRanges()).containsExactly(new Range(1, 2));
IndexedSplit firstSplit = (IndexedSplit) result.splits().get(1);
assertThat(firstSplit.dataSplit().dataFiles()).containsExactly(first);
assertThat(firstSplit.rowRanges()).containsExactly(new Range(1, 1));
+ IndexedSplit secondSplit = (IndexedSplit) result.splits().get(2);
+
assertThat(secondSplit.dataSplit().dataFiles()).containsExactly(second);
+ assertThat(secondSplit.rowRanges()).containsExactly(new Range(0, 0),
new Range(2, 2));
}
@Test
