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 3b7d606ac8 [core] Parallelize primary-key full-text search (#9184)
3b7d606ac8 is described below
commit 3b7d606ac863ecccfec4a4b3117cee005e3f70c1
Author: QuakeWang <[email protected]>
AuthorDate: Wed Aug 12 14:10:10 2026 +0800
[core] Parallelize primary-key full-text search (#9184)
---
.../pkfulltext/PrimaryKeyFullTextBucketSearch.java | 17 +++-
.../table/source/PrimaryKeyFullTextRead.java | 24 +++--
.../PrimaryKeyFullTextBucketSearchTest.java | 61 +++++++++++
.../table/source/PrimaryKeyFullTextReadTest.java | 113 ++++++++++++++++++---
4 files changed, 192 insertions(+), 23 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
index 2bdebce47d..89719c48c2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
@@ -75,6 +75,15 @@ public class PrimaryKeyFullTextBucketSearch {
String column,
String query,
int limit) {
+ return searchRankingsAsync(split, deletionVectors, column, query,
limit).join();
+ }
+
+ public CompletableFuture<List<List<PrimaryKeySearchPosition>>>
searchRankingsAsync(
+ PrimaryKeyFullTextSearchSplit split,
+ Map<String, DeletionVector> deletionVectors,
+ String column,
+ String query,
+ int limit) {
checkArgument(limit > 0, "Full-text search limit must be positive:
%s.", limit);
DataSplit dataSplit = split.dataSplit();
Map<String, DataFileMeta> files = new HashMap<>();
@@ -130,11 +139,15 @@ public class PrimaryKeyFullTextBucketSearch {
requests.add(new PayloadRequest(sourceRanges, totalRowCount,
include, future));
}
- CompletableFuture.allOf(
+ return CompletableFuture.allOf(
requests.stream()
.map(request -> request.future)
.toArray(CompletableFuture[]::new))
- .join();
+ .thenApply(ignored -> collectRankings(dataSplit, requests));
+ }
+
+ private static List<List<PrimaryKeySearchPosition>> collectRankings(
+ DataSplit dataSplit, List<PayloadRequest> requests) {
List<List<PrimaryKeySearchPosition>> localRankings = new
ArrayList<>(requests.size());
for (PayloadRequest request : requests) {
Optional<ScoredGlobalIndexResult> result = request.future.join();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java
index edc56a372f..24a823731a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java
@@ -43,6 +43,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
@@ -73,7 +74,7 @@ public class PrimaryKeyFullTextRead implements FullTextRead {
ProductionSearch production =
new ProductionSearch(table, definition, textField, query,
limit);
this.limit = limit;
- this.indexedSearch = production::searchIndexed;
+ this.indexedSearch = production::searchIndexedAsync;
}
PrimaryKeyFullTextRead(
@@ -115,7 +116,10 @@ public class PrimaryKeyFullTextRead implements
FullTextRead {
private PrimaryKeyScoredResult read(long snapshotId,
List<FullTextSearchSplit> splits) {
List<DataSplit> sourceSplits = new ArrayList<>(splits.size());
- List<List<PrimaryKeySearchPosition>> rankings = new ArrayList<>();
+ List<CompletableFuture<List<List<PrimaryKeySearchPosition>>>> futures =
+ new ArrayList<>(splits.size());
+ // Start from the caller because each bucket submits payload searches
to the same executor.
+ // Wrapping the whole bucket in that executor and waiting inside it
can starve leaf work.
for (FullTextSearchSplit searchSplit : splits) {
checkArgument(
searchSplit instanceof PrimaryKeyFullTextSearchSplit,
@@ -125,7 +129,12 @@ public class PrimaryKeyFullTextRead implements
FullTextRead {
split.dataSplit().snapshotId() == snapshotId,
"Full-text bucket split snapshot does not match its
plan.");
sourceSplits.add(split.dataSplit());
- rankings.addAll(indexedSearch.search(split));
+ futures.add(indexedSearch.searchAsync(split));
+ }
+ CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).join();
+ List<List<PrimaryKeySearchPosition>> rankings = new ArrayList<>();
+ for (CompletableFuture<List<List<PrimaryKeySearchPosition>>> future :
futures) {
+ rankings.addAll(future.join());
}
List<PrimaryKeySearchPosition> positions =
rankings.isEmpty()
@@ -136,7 +145,8 @@ public class PrimaryKeyFullTextRead implements FullTextRead
{
@FunctionalInterface
interface BucketRankingSearch {
- List<List<PrimaryKeySearchPosition>>
search(PrimaryKeyFullTextSearchSplit split);
+ CompletableFuture<List<List<PrimaryKeySearchPosition>>> searchAsync(
+ PrimaryKeyFullTextSearchSplit split);
}
private static class ProductionSearch {
@@ -170,13 +180,13 @@ public class PrimaryKeyFullTextRead implements
FullTextRead {
this.archiveReader = meta ->
fileIO.newInputStream(meta.filePath());
}
- private List<List<PrimaryKeySearchPosition>> searchIndexed(
+ private CompletableFuture<List<List<PrimaryKeySearchPosition>>>
searchIndexedAsync(
PrimaryKeyFullTextSearchSplit split) {
if (split.payloadFiles().isEmpty()) {
- return Collections.emptyList();
+ return
CompletableFuture.completedFuture(Collections.emptyList());
}
return bucketSearch(split)
- .searchRankings(
+ .searchRankingsAsync(
split,
deletionVectors(split.dataSplit()),
textField.name(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
index 231343cac5..5e0ac8a60b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
@@ -51,6 +51,11 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+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;
@@ -61,6 +66,62 @@ import static
org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
/** Tests payload-local full-text search and cross-payload score merging. */
class PrimaryKeyFullTextBucketSearchTest {
+ @Test
+ void testStartsPayloadsBeforeWaitingForResults() throws Exception {
+ CompletableFuture<Optional<ScoredGlobalIndexResult>> futureA = new
CompletableFuture<>();
+ CompletableFuture<Optional<ScoredGlobalIndexResult>> futureB = new
CompletableFuture<>();
+ CountDownLatch started = new CountDownLatch(2);
+ AtomicInteger closes = new AtomicInteger();
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ (payload, ignoredTotalRowCount) -> {
+ String source =
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
+ .sourceFile()
+ .fileName();
+ return new FullTextOnlyReader() {
+ @Override
+ public
CompletableFuture<Optional<ScoredGlobalIndexResult>>
+ visitFullTextSearch(FullTextSearch
fullTextSearch) {
+ started.countDown();
+ return source.equals("a") ? futureA :
futureB;
+ }
+
+ @Override
+ public void close() {
+ closes.incrementAndGet();
+ }
+ };
+ });
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ try {
+ Future<CompletableFuture<List<List<PrimaryKeySearchPosition>>>>
invocation =
+ executor.submit(
+ () ->
+ search.searchRankingsAsync(
+ split(),
+ Collections.emptyMap(),
+ "content",
+ "hello",
+ 2));
+ assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+ CompletableFuture<List<List<PrimaryKeySearchPosition>>>
resultFuture =
+ invocation.get(5, TimeUnit.SECONDS);
+ assertThat(resultFuture.isDone()).isFalse();
+ futureB.complete(Optional.empty());
+ assertThat(resultFuture.isDone()).isFalse();
+ futureA.complete(Optional.empty());
+ assertThat(resultFuture.get(5, TimeUnit.SECONDS)).isEmpty();
+ } finally {
+ futureA.complete(Optional.empty());
+ futureB.complete(Optional.empty());
+ executor.shutdownNow();
+ }
+ assertThat(closes).hasValue(2);
+ }
+
@Test
void testFiltersDeletedRowsAndSelectsGlobalScores() {
PrimaryKeyFullTextSearchSplit split = split();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
index a791a605b2..a9391c1f5b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
@@ -33,9 +33,17 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+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 static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -44,6 +52,64 @@ import static org.assertj.core.api.Assertions.tuple;
/** Tests PK full-text search modes and physical scored results. */
class PrimaryKeyFullTextReadTest {
+ @Test
+ void testStartsBucketsBeforeWaitingForResults() throws Exception {
+ PrimaryKeyFullTextSearchSplit split0 = split("indexed-0", "raw-0", 0,
2);
+ PrimaryKeyFullTextSearchSplit split1 = split("indexed-1", "raw-1", 1,
2);
+ CompletableFuture<List<List<PrimaryKeySearchPosition>>> future0 = new
CompletableFuture<>();
+ CompletableFuture<List<List<PrimaryKeySearchPosition>>> future1 = new
CompletableFuture<>();
+ CountDownLatch started = new CountDownLatch(2);
+ PrimaryKeyFullTextRead read =
+ new PrimaryKeyFullTextRead(
+ GlobalIndexSearchMode.FAST,
+ 10,
+ split -> {
+ started.countDown();
+ if (split == split0) {
+ return future0;
+ }
+ assertThat(split).isSameAs(split1);
+ return future1;
+ });
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ try {
+ Future<PrimaryKeyScoredResult> resultFuture =
+ executor.submit(
+ () ->
read.read(Arrays.<FullTextSearchSplit>asList(split0, split1)));
+ assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+
+ future1.complete(
+ Collections.singletonList(
+ Collections.singletonList(position("indexed-1", 1,
0, 9F))));
+ future0.complete(
+ Collections.singletonList(
+ Collections.singletonList(position("indexed-0", 0,
0, 9F))));
+
+ assertThat(resultFuture.get(5, TimeUnit.SECONDS).positions())
+ .extracting(
+ PrimaryKeySearchPosition::bucket,
+ PrimaryKeySearchPosition::dataFileName)
+ .containsExactly(tuple(0, "indexed-0"), tuple(1,
"indexed-1"));
+ } finally {
+ future0.completeExceptionally(new RuntimeException("Test
cleanup."));
+ future1.completeExceptionally(new RuntimeException("Test
cleanup."));
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testPropagatesBucketFailure() {
+ CompletableFuture<List<List<PrimaryKeySearchPosition>>> failed = new
CompletableFuture<>();
+ failed.completeExceptionally(new IOException("broken"));
+ PrimaryKeyFullTextRead read =
+ new PrimaryKeyFullTextRead(GlobalIndexSearchMode.FAST, 10,
split -> failed);
+
+ assertThatThrownBy(() -> read.read(Collections.singletonList(split())))
+ .isInstanceOf(CompletionException.class)
+ .hasRootCauseMessage("broken");
+ }
+
@Test
void testFastPropagatesScores() {
PrimaryKeyFullTextRead read =
@@ -51,8 +117,10 @@ class PrimaryKeyFullTextReadTest {
GlobalIndexSearchMode.FAST,
10,
split ->
- Collections.singletonList(
-
Collections.singletonList(position("indexed", 1, 9F))));
+ CompletableFuture.completedFuture(
+ Collections.singletonList(
+ Collections.singletonList(
+ position("indexed", 1,
9F)))));
PrimaryKeyScoredResult result =
read.read(Collections.singletonList(split()));
@@ -74,11 +142,13 @@ class PrimaryKeyFullTextReadTest {
GlobalIndexSearchMode.FAST,
2,
split ->
- Arrays.asList(
+ CompletableFuture.completedFuture(
Arrays.asList(
- position("indexed", 0, 100F),
- position("indexed", 1, 99F)),
-
Collections.singletonList(position("raw", 0, 1F))));
+ Arrays.asList(
+ position("indexed", 0,
100F),
+ position("indexed", 1,
99F)),
+ Collections.singletonList(
+ position("raw", 0,
1F)))));
PrimaryKeyScoredResult result =
read.read(Collections.singletonList(split()));
@@ -98,31 +168,46 @@ class PrimaryKeyFullTextReadTest {
assertThatThrownBy(
() ->
new PrimaryKeyFullTextRead(
- mode, 10, split ->
Collections.emptyList()))
+ mode,
+ 10,
+ split ->
+
CompletableFuture.completedFuture(
+
Collections.emptyList())))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("only supports the FAST full-text-index
search mode");
}
private static PrimaryKeySearchPosition position(
String dataFile, long rowPosition, float score) {
- return new PrimaryKeySearchPosition(BinaryRow.EMPTY_ROW, 0, dataFile,
rowPosition, score);
+ return position(dataFile, 0, rowPosition, score);
+ }
+
+ private static PrimaryKeySearchPosition position(
+ String dataFile, int bucket, long rowPosition, float score) {
+ return new PrimaryKeySearchPosition(
+ BinaryRow.EMPTY_ROW, bucket, dataFile, rowPosition, score);
}
private static PrimaryKeyFullTextSearchSplit split() {
- List<DataFileMeta> dataFiles = Arrays.asList(dataFile("indexed"),
dataFile("raw"));
+ return split("indexed", "raw", 0, 1);
+ }
+
+ private static PrimaryKeyFullTextSearchSplit split(
+ String indexedFile, String rawFile, int bucket, int totalBuckets) {
+ List<DataFileMeta> dataFiles = Arrays.asList(dataFile(indexedFile),
dataFile(rawFile));
DataSplit dataSplit =
DataSplit.builder()
.withSnapshot(11)
.withPartition(BinaryRow.EMPTY_ROW)
- .withBucket(0)
- .withBucketPath("bucket-0")
- .withTotalBuckets(1)
+ .withBucket(bucket)
+ .withBucketPath("bucket-" + bucket)
+ .withTotalBuckets(totalBuckets)
.withDataFiles(dataFiles)
.build();
return new PrimaryKeyFullTextSearchSplit(
dataSplit,
- Collections.singletonList(payload("indexed")),
- Collections.singletonList("raw"));
+ Collections.singletonList(payload(indexedFile)),
+ Collections.singletonList(rawFile));
}
private static DataFileMeta dataFile(String name) {
