This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit e4f49ec969fc5acd4f86b4e5cefe8e0294490857 Author: Xiangyi Zhu <[email protected]> AuthorDate: Fri Sep 18 19:33:51 2026 +0800 [core] Request only top-k candidates from data-evolution full-text indexes (#9955) --- .../TestFullTextGlobalIndexReader.java | 5 ++ .../table/source/DataEvolutionFullTextRead.java | 13 ++-- .../table/source/FullTextSearchBuilderTest.java | 70 ++++++++++++++++++++++ 3 files changed, 79 insertions(+), 9 deletions(-) diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java index b2cd4ca23e..c695c90950 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java @@ -43,6 +43,7 @@ import java.util.Map; import java.util.Optional; import java.util.PriorityQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; /** * Test full-text index reader that performs brute-force text matching. Loads all documents into @@ -60,6 +61,9 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader { private long[] rowIds; private int count; + /** Limits requested from this reader, in call order; lets tests pin what the read asks for. */ + public static final List<Integer> REQUESTED_LIMITS = new CopyOnWriteArrayList<>(); + public TestFullTextGlobalIndexReader( GlobalIndexFileReader fileReader, GlobalIndexIOMeta ioMeta) { this.fileReader = fileReader; @@ -76,6 +80,7 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader { } int limit = fullTextSearch.limit(); + REQUESTED_LIMITS.add(limit); int effectiveK = Math.min(limit, count); if (effectiveK <= 0) { return CompletableFuture.completedFuture(Optional.empty()); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java index 9a8227b1b9..8704145e25 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java @@ -451,19 +451,14 @@ public class DataEvolutionFullTextRead implements FullTextRead { indexIOMetaList, rowRangeEnd - rowRangeStart + 1, executor); + // Each split returns its own top-k; the union's top-k after merge equals the global + // top-k, and the native engine scores compound queries per document, so there is no + // need to request every candidate of the range. FullTextSearch fullTextSearch = - new FullTextSearch( - textColumn.name(), - query, - candidateLimit(rowRangeStart, rowRangeEnd)) + new FullTextSearch(textColumn.name(), query, limit) .withIncludeRowIds(includeRowIds); return new OffsetGlobalIndexReader(reader, rowRangeStart, rowRangeEnd) .visitFullTextSearch(fullTextSearch) .whenComplete((r, t) -> IOUtils.closeQuietly(reader)); } - - private static int candidateLimit(long rowRangeStart, long rowRangeEnd) { - long size = rowRangeEnd - rowRangeStart + 1; - return size > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) size; - } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java index 94f483ec6c..388d05700e 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java @@ -31,6 +31,7 @@ import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; import org.apache.paimon.globalindex.ResultEntry; import org.apache.paimon.globalindex.ScoredGlobalIndexResult; import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory; +import org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexReader; import org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileMeta; @@ -1045,6 +1046,75 @@ public class FullTextSearchBuilderTest extends TableTestBase { assertThat(ids.size()).isLessThanOrEqualTo(5); } + @Test + public void testFullTextSearchRequestsOnlyLimitCandidatesPerSplit() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + writeDocuments(table, RANKED_DOCUMENTS); + List<DataField> textFields = + Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)); + buildAndCommitIndexRange(table, Arrays.copyOfRange(RANKED_DOCUMENTS, 0, 3), textFields, 0); + buildAndCommitIndexRange(table, Arrays.copyOfRange(RANKED_DOCUMENTS, 3, 6), textFields, 3); + + TestFullTextGlobalIndexReader.REQUESTED_LIMITS.clear(); + GlobalIndexResult result = + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon")) + .withLimit(2) + .executeLocal(); + + // Each of the two splits (3 rows each) is asked for the user limit, not its row count. + assertThat(TestFullTextGlobalIndexReader.REQUESTED_LIMITS).containsExactly(2, 2); + assertThat(result.results()).hasSize(2); + } + + @Test + public void testFullTextSearchMergesPerSplitTopKIntoGlobalTopK() throws Exception { + createTableDefault(); + FileStoreTable table = getTableDefault(); + // Split 1 (rows 0-2) holds only 0.5-scored matches; split 2 (rows 3-5) holds the two + // 1.0-scored matches plus one 0.5. A global top-2 must come entirely from split 2. + String[] documents = { + "paimon alpha", "paimon beta", "paimon gamma", + "paimon lake delta", "paimon lake epsilon", "paimon zeta" + }; + writeDocuments(table, documents); + List<DataField> textFields = + Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)); + buildAndCommitIndexRange(table, Arrays.copyOfRange(documents, 0, 3), textFields, 0); + buildAndCommitIndexRange(table, Arrays.copyOfRange(documents, 3, 6), textFields, 3); + + ScoredGlobalIndexResult top2 = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon lake")) + .withLimit(2) + .executeLocal(); + assertThat(top2.results()).containsExactlyInAnyOrder(3L, 4L); + + ScoredGlobalIndexResult top1 = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon lake")) + .withLimit(1) + .executeLocal(); + assertThat(top1.results()).hasSize(1); + assertThat(top1.results()).isSubsetOf(3L, 4L); + + // A larger limit still ranks the 1.0 rows first and fills up with 0.5 rows. + ScoredGlobalIndexResult top4 = + (ScoredGlobalIndexResult) + table.newFullTextSearchBuilder() + .withQuery(TEXT_FIELD_NAME, matchQuery("paimon lake")) + .withLimit(4) + .executeLocal(); + assertThat(top4.results()).hasSize(4); + assertThat(top4.results()).contains(3L, 4L); + for (long rowId : top4.results()) { + assertThat(top4.scoreGetter().score(rowId)).isGreaterThanOrEqualTo(0.5f); + } + } + @Test public void testFullTextSearchWithMultipleIndexFiles() throws Exception { createTableDefault();
