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 fec6be9e8a [core] Parallelize primary-key sorted-index scan (#9231)
fec6be9e8a is described below
commit fec6be9e8ac5f26cf224bf1950384c81a00005c2
Author: QuakeWang <[email protected]>
AuthorDate: Sat Aug 15 20:12:08 2026 +0800
[core] Parallelize primary-key sorted-index scan (#9231)
---
.../paimon/globalindex/GlobalIndexEvaluator.java | 12 ++-
.../globalindex/GlobalIndexEvaluatorTest.java | 27 +++++++
.../table/source/PrimaryKeySortedIndexScan.java | 38 +++++++++-
.../source/PrimaryKeySortedIndexScanTest.java | 85 ++++++++++++++++++++++
4 files changed, 157 insertions(+), 5 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java
index a23bd2c76a..f4a55cfbec 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java
@@ -69,10 +69,18 @@ public class GlobalIndexEvaluator implements Closeable {
}
public Optional<GlobalIndexResult> evaluate(@Nullable Predicate predicate)
{
+ return await(evaluateAsync(predicate));
+ }
+
+ /**
+ * Evaluate the predicate asynchronously. Keep this evaluator open until
the future completes.
+ */
+ public CompletableFuture<Optional<GlobalIndexResult>> evaluateAsync(
+ @Nullable Predicate predicate) {
if (predicate == null) {
- return Optional.empty();
+ return CompletableFuture.completedFuture(Optional.empty());
}
- return await(visitAsync(predicate)).map(Evaluation::result);
+ return visitAsync(predicate).thenApply(result ->
result.map(Evaluation::result));
}
/** Evaluate the predicate and return the fields whose supported indexes
contributed. */
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
index 8b7d677ed3..ec193fd5e9 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java
@@ -101,6 +101,33 @@ class GlobalIndexEvaluatorTest {
evaluator.close();
}
+ @Test
+ void testEvaluateAsyncDoesNotWaitForReaderResult() {
+ RowType rowType = rowType();
+ CompletableFuture<Optional<GlobalIndexResult>> readerResult = new
CompletableFuture<>();
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(
+ rowType,
+ fieldId ->
+ Collections.singletonList(
+ new StubGlobalIndexReader(null) {
+ @Override
+ public
CompletableFuture<Optional<GlobalIndexResult>>
+ visitEqual(FieldRef
fieldRef, Object literal) {
+ return readerResult;
+ }
+ }));
+ Predicate predicate = new PredicateBuilder(rowType).equal(0, 42);
+
+ CompletableFuture<Optional<GlobalIndexResult>> result =
evaluator.evaluateAsync(predicate);
+
+ assertThat(result.isDone()).isFalse();
+ readerResult.complete(Optional.of(resultOf(1, 2, 3)));
+ assertThat(result.join()).isPresent();
+ assertBitmapContainsExactly(result.join().get().results(), 1L, 2L, 3L);
+ evaluator.close();
+ }
+
@Test
void testTopNUsesAggregatedReaderAndReusesPredicateCache() {
RowType rowType = rowType();
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 cb03736b7b..b1458eadbc 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
@@ -42,6 +42,7 @@ import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.FutureUtils;
import org.apache.paimon.utils.IOUtils;
import org.apache.paimon.utils.IndexFilePathFactories;
import org.apache.paimon.utils.Pair;
@@ -65,6 +66,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;
@@ -246,6 +248,8 @@ public final class PrimaryKeySortedIndexScan {
}
Map<PkSortedIndexGroup, SharedGlobalIndexReader> sharedReaders = new
IdentityHashMap<>();
+ List<GlobalIndexEvaluator> evaluators = new ArrayList<>();
+ List<CompletableFuture<Optional<GlobalIndexResult>>> resultFutures =
new ArrayList<>();
List<EvaluatedFile> files = new ArrayList<>();
try {
for (FilePlan file : plan.files()) {
@@ -277,9 +281,20 @@ public final class PrimaryKeySortedIndexScan {
}
return
Collections.singletonList(fileLocalReader(file, reader));
});
+ evaluators.add(evaluator);
+ try {
+ resultFutures.add(evaluator.evaluateAsync(predicate));
+ } catch (RuntimeException e) {
+ rethrowIfInterrupted(e);
+ resultFutures.add(FutureUtils.completedExceptionally(e));
+ }
+ }
+
+ for (int i = 0; i < plan.files().size(); i++) {
+ FilePlan file = plan.files().get(i);
Optional<GlobalIndexResult> result;
try {
- result = evaluator.evaluate(predicate);
+ result = awaitEvaluation(resultFutures.get(i));
} catch (RuntimeException e) {
rethrowIfInterrupted(e);
LOG.warn(
@@ -288,17 +303,34 @@ public final class PrimaryKeySortedIndexScan {
file.dataFile().fileName(),
e);
result = Optional.empty();
- } finally {
- evaluator.close();
}
files.add(new EvaluatedFile(file, result));
}
} finally {
+ IOUtils.closeAllQuietly(evaluators);
IOUtils.closeAllQuietly(sharedReaders.values());
}
return new EvaluatedPlan(plan.snapshotId(), files);
}
+ private static Optional<GlobalIndexResult> awaitEvaluation(
+ CompletableFuture<Optional<GlobalIndexResult>> future) {
+ try {
+ return future.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted during index evaluation",
e);
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof RuntimeException) {
+ throw (RuntimeException) e.getCause();
+ }
+ if (e.getCause() instanceof Error) {
+ throw (Error) e.getCause();
+ }
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
private static GlobalIndexReader fileLocalReader(
FilePlan file, SharedGlobalIndexReader reader) {
DataFileMeta dataFile = file.dataFile();
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 b800fb1c61..9281ec395a 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
@@ -54,6 +54,11 @@ import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.assertj.core.api.Assertions.assertThat;
@@ -69,6 +74,86 @@ import static org.mockito.Mockito.when;
/** Tests source-backed BTree and Bitmap planning in file-local row-position
space. */
class PrimaryKeySortedIndexScanTest {
+ @Test
+ void testStartsIndependentGroupsBeforeWaitingForResults() throws Exception
{
+ DataSplit firstSplit = dataSplit(11, 0, dataFile("data-1", 4));
+ DataSplit secondSplit = dataSplit(11, 1, dataFile("data-2", 4));
+ PrimaryKeyIndexDefinition definition =
+ definition(
+ 7,
+ BTreeGlobalIndexerFactory.IDENTIFIER,
+ PrimaryKeyIndexDefinition.Family.BTREE);
+ PrimaryKeySortedIndexScan.Plan plan =
+ PrimaryKeySortedIndexScan.plan(
+ 11,
+ Arrays.asList(firstSplit, secondSplit),
+ Collections.singletonList(definition),
+ Arrays.asList(
+ payloadEntry(0, payload("btree-0", "data-1",
4, "btree", 7, 4)),
+ payloadEntry(1, payload("btree-1", "data-2",
4, "btree", 7, 4))));
+ assertThat(plan.files()).hasSize(2);
+ assertThat(plan.files()).allSatisfy(file ->
assertThat(file.group(7)).isPresent());
+
+ RowType rowType = RowType.of(new DataField(7, "f7", DataTypes.INT()));
+ Predicate predicate = new PredicateBuilder(rowType).equal(0, 42);
+ CompletableFuture<Optional<GlobalIndexResult>> firstResult = new
CompletableFuture<>();
+ CompletableFuture<Optional<GlobalIndexResult>> secondResult = new
CompletableFuture<>();
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ GlobalIndexReader firstReader = mock(GlobalIndexReader.class);
+ when(firstReader.visitEqual(any(), eq(42)))
+ .thenAnswer(
+ ignored -> {
+ firstStarted.countDown();
+ return firstResult;
+ });
+ GlobalIndexReader secondReader = mock(GlobalIndexReader.class);
+ when(secondReader.visitEqual(any(), eq(42)))
+ .thenAnswer(
+ ignored -> {
+ secondStarted.countDown();
+ return secondResult;
+ });
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ try {
+ Future<PrimaryKeySortedIndexScan.EvaluatedPlan> evaluated =
+ executor.submit(
+ () ->
+ PrimaryKeySortedIndexScan.evaluate(
+ plan,
+ rowType,
+ predicate,
+
Collections.singletonList(definition),
+ (file,
+ ignoredDefinition,
+ ignoredPayloads,
+ ignoredTotalRowCount) ->
+
file.dataFile().fileName().equals("data-1")
+ ? firstReader
+ : secondReader));
+
+ assertThat(firstStarted.await(5, TimeUnit.SECONDS)).isTrue();
+ boolean secondStartedBeforeFirstCompleted = secondStarted.await(1,
TimeUnit.SECONDS);
+ firstResult.complete(Optional.of(GlobalIndexResult.createEmpty()));
+ assertThat(secondStarted.await(5, TimeUnit.SECONDS))
+ .as("the second index group should eventually start")
+ .isTrue();
+
secondResult.complete(Optional.of(GlobalIndexResult.createEmpty()));
+ assertThat(evaluated.get(5, TimeUnit.SECONDS).files()).hasSize(2);
+ verify(firstReader).close();
+ verify(secondReader).close();
+
+ assertThat(secondStartedBeforeFirstCompleted)
+ .as("the second index group should start before the first
result completes")
+ .isTrue();
+ } finally {
+ firstResult.completeExceptionally(new RuntimeException("Test
cleanup."));
+ secondResult.completeExceptionally(new RuntimeException("Test
cleanup."));
+ executor.shutdownNow();
+ }
+ }
+
@Test
void testPayloadStateIsBuiltOncePerBucketAndDefinition() {
DataSplit split =
