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 12b9facba0 [core] Centralize negative predicate complements in sorted
index readers (#9060)
12b9facba0 is described below
commit 12b9facba064600de251beca0d03950a39e918e3
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Aug 6 16:24:57 2026 +0800
[core] Centralize negative predicate complements in sorted index readers
(#9060)
---
.../paimon/globalindex/GlobalIndexReader.java | 5 --
.../apache/paimon/globalindex/GlobalIndexer.java | 7 ++
.../globalindex/OffsetGlobalIndexReader.java | 39 ----------
.../globalindex/SortedFileGlobalIndexReader.java | 56 +++++++++++----
.../globalindex/bitmap/BitmapGlobalIndexer.java | 3 +-
.../bitmap/LazyFilteredBitmapReader.java | 19 +----
.../globalindex/btree/BTreeGlobalIndexer.java | 2 +
.../globalindex/btree/LazyFilteredBTreeReader.java | 24 +------
.../globalindex/GlobalIndexEvaluatorTest.java | 82 +++++-----------------
.../bitmap/LazyFilteredBitmapIndexReaderTest.java | 72 +++++++++++++++++--
.../globalindex/btree/BTreeIndexReaderTest.java | 5 +-
.../globalindex/btree/BTreeThreadSafetyTest.java | 13 ++--
.../btree/LazyFilteredBTreeIndexReaderTest.java | 17 +++--
.../testfulltext/TestFullTextGlobalIndexer.java | 1 +
.../TestMultiFieldVectorGlobalIndexer.java | 5 +-
.../testvector/TestVectorGlobalIndexer.java | 1 +
.../DataEvolutionGlobalIndexScanner.java | 5 +-
.../pkfulltext/PrimaryKeyFullTextBucketSearch.java | 4 +-
.../index/pkvector/PkVectorAnnSegmentSearcher.java | 2 +
.../source/AbstractDataEvolutionVectorRead.java | 12 +++-
.../table/source/DataEvolutionFullTextRead.java | 6 +-
.../table/source/PrimaryKeyFullTextRead.java | 7 +-
.../table/source/PrimaryKeySortedIndexScan.java | 27 +++++--
.../index/pkfulltext/PkFullTextIndexFileTest.java | 1 +
.../PrimaryKeyFullTextBucketSearchTest.java | 13 ++--
.../index/pksorted/PkSortedIndexBuilderTest.java | 1 +
.../source/PrimaryKeySortedIndexBatchScanTest.java | 3 +-
.../source/PrimaryKeySortedIndexResultTest.java | 7 +-
.../source/PrimaryKeySortedIndexScanTest.java | 14 ++--
.../paimon/eslib/index/ESIndexGlobalIndexer.java | 1 +
.../index/ESIndexGlobalIndexerExecutorTest.java | 4 +-
.../eslib/index/ESIndexVectorMetricTest.java | 4 +-
.../FlinkDataEvolutionVectorReadTest.java | 1 +
.../index/NativeFullTextGlobalIndexer.java | 1 +
.../index/NativeFullTextGlobalIndexTest.java | 2 +-
.../index/NativePrimaryKeyFullTextIndexTest.java | 3 +-
.../lumina/index/LuminaVectorGlobalIndexer.java | 1 +
.../read/SparkDataEvolutionVectorReadTest.java | 1 +
.../vector/index/NativeVectorGlobalIndexer.java | 1 +
.../vector/index/NativeVectorGlobalIndexTest.java | 3 +-
40 files changed, 256 insertions(+), 219 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
index 694d367aaa..b857fdc1a1 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java
@@ -36,11 +36,6 @@ import java.util.concurrent.CompletableFuture;
public interface GlobalIndexReader
extends
FunctionVisitor<CompletableFuture<Optional<GlobalIndexResult>>>, Closeable {
- /** Whether this reader can answer negative predicates by complementing a
known row range. */
- default boolean supportsRangeComplement() {
- return false;
- }
-
@Override
default CompletableFuture<Optional<GlobalIndexResult>> visitIsNaN(FieldRef
fieldRef) {
return CompletableFuture.completedFuture(Optional.empty());
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
index 5eadf0597f..eade369f7b 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexer.java
@@ -32,9 +32,16 @@ public interface GlobalIndexer {
GlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter) throws
IOException;
+ /**
+ * Creates a reader whose relative row IDs cover {@code [0,
totalRowCount)}.
+ *
+ * <p>The complete row count lets an index implementation answer negative
predicates by
+ * complement, including for an empty shard.
+ */
GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor);
static GlobalIndexer create(String type, DataField indexField, Options
options) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
index 064893bf2e..23d38d3a94 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java
@@ -23,7 +23,6 @@ import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FullTextSearch;
import org.apache.paimon.predicate.TopN;
import org.apache.paimon.predicate.VectorSearch;
-import org.apache.paimon.utils.Range;
import java.io.IOException;
import java.util.ArrayList;
@@ -49,9 +48,6 @@ public class OffsetGlobalIndexReader implements
GlobalIndexReader {
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
- if (wrapped.supportsRangeComplement()) {
- return complement(wrapped.visitIsNull(fieldRef));
- }
return wrapped.visitIsNotNull(fieldRef).thenApply(this::applyOffset);
}
@@ -99,12 +95,6 @@ public class OffsetGlobalIndexReader implements
GlobalIndexReader {
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
FieldRef fieldRef, Object literal) {
- if (literal == null) {
- return
CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.createEmpty()));
- }
- if (wrapped.supportsRangeComplement()) {
- return complement(wrapped.visitIsNull(fieldRef),
wrapped.visitEqual(fieldRef, literal));
- }
return wrapped.visitNotEqual(fieldRef,
literal).thenApply(this::applyOffset);
}
@@ -135,15 +125,6 @@ public class OffsetGlobalIndexReader implements
GlobalIndexReader {
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
FieldRef fieldRef, List<Object> literals) {
- for (Object literal : literals) {
- if (literal == null) {
- return CompletableFuture.completedFuture(
- Optional.of(GlobalIndexResult.createEmpty()));
- }
- }
- if (wrapped.supportsRangeComplement()) {
- return complement(wrapped.visitIsNull(fieldRef),
wrapped.visitIn(fieldRef, literals));
- }
return wrapped.visitNotIn(fieldRef,
literals).thenApply(this::applyOffset);
}
@@ -197,26 +178,6 @@ public class OffsetGlobalIndexReader implements
GlobalIndexReader {
return result.map(r -> r.offset(offset));
}
- @SafeVarargs
- private final CompletableFuture<Optional<GlobalIndexResult>> complement(
- CompletableFuture<Optional<GlobalIndexResult>>... excludeFutures) {
- return CompletableFuture.allOf(excludeFutures)
- .thenApply(
- ignored -> {
- GlobalIndexResult result =
- GlobalIndexResult.fromRange(new
Range(offset, to));
- for
(CompletableFuture<Optional<GlobalIndexResult>> future :
- excludeFutures) {
- Optional<GlobalIndexResult> excluded =
future.join();
- if (!excluded.isPresent()) {
- return Optional.empty();
- }
- result =
result.andNot(excluded.get().offset(offset));
- }
- return Optional.of(result);
- });
- }
-
@Override
public void close() throws IOException {
wrapped.close();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java
index f16c2036fe..814ec92414 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java
@@ -29,6 +29,7 @@ import org.apache.paimon.predicate.LikeOptimization;
import org.apache.paimon.predicate.StartsWith;
import org.apache.paimon.types.DataTypeFamily;
import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import java.io.Closeable;
@@ -53,22 +54,29 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
private final long fallbackScanMaxSize;
private final Map<Path, R> readerCache;
private final ExecutorService executor;
+ private final long totalRowCount;
protected SortedFileGlobalIndexReader(
List<GlobalIndexIOMeta> files,
KeySerializer keySerializer,
long fallbackScanMaxSize,
+ long totalRowCount,
ExecutorService executor) {
+ if (totalRowCount < 0) {
+ throw new IllegalArgumentException(
+ "Total row count must be non-negative, but was " +
totalRowCount + ".");
+ }
this.fileSelector = new SortedFileMetaSelector(files, keySerializer);
this.files = new ArrayList<>(files);
this.fallbackScanMaxSize = fallbackScanMaxSize;
this.readerCache = new ConcurrentHashMap<>();
this.executor = executor;
+ this.totalRowCount = totalRowCount;
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
- return visitParallel(() -> fileSelector.visitIsNotNull(fieldRef),
this::visitIsNotNull);
+ return complement(visitIsNull(fieldRef));
}
@Override
@@ -169,9 +177,10 @@ public abstract class SortedFileGlobalIndexReader<R
extends Closeable>
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
FieldRef fieldRef, Object literal) {
- return visitParallel(
- () -> fileSelector.visitNotEqual(fieldRef, literal),
- reader -> visitNotEqual(reader, literal));
+ if (literal == null) {
+ return
CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.createEmpty()));
+ }
+ return complement(visitIsNull(fieldRef), visitEqual(fieldRef,
literal));
}
@Override
@@ -215,9 +224,13 @@ public abstract class SortedFileGlobalIndexReader<R
extends Closeable>
@Override
public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
FieldRef fieldRef, List<Object> literals) {
- return visitParallel(
- () -> fileSelector.visitNotIn(fieldRef, literals),
- reader -> visitNotIn(reader, literals));
+ for (Object literal : literals) {
+ if (literal == null) {
+ return CompletableFuture.completedFuture(
+ Optional.of(GlobalIndexResult.createEmpty()));
+ }
+ }
+ return complement(visitIsNull(fieldRef), visitIn(fieldRef, literals));
}
@Override
@@ -246,8 +259,6 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
reader -> visitNotBetween(reader, from, to));
}
- protected abstract Optional<GlobalIndexResult> visitIsNotNull(R reader);
-
protected abstract Optional<GlobalIndexResult> visitIsNull(R reader);
protected abstract Optional<GlobalIndexResult> visitStartsWith(R reader,
Object literal);
@@ -260,8 +271,6 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
protected abstract Optional<GlobalIndexResult> visitGreaterOrEqual(R
reader, Object literal);
- protected abstract Optional<GlobalIndexResult> visitNotEqual(R reader,
Object literal);
-
protected abstract Optional<GlobalIndexResult> visitLessOrEqual(R reader,
Object literal);
protected abstract Optional<GlobalIndexResult> visitEqual(R reader, Object
literal);
@@ -270,8 +279,6 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
protected abstract Optional<GlobalIndexResult> visitIn(R reader,
List<Object> literals);
- protected abstract Optional<GlobalIndexResult> visitNotIn(R reader,
List<Object> literals);
-
protected abstract Optional<GlobalIndexResult> visitBetween(R reader,
Object from, Object to);
protected Optional<GlobalIndexResult> visitLike(R reader, FieldRef
fieldRef, Object literal) {
@@ -348,6 +355,29 @@ public abstract class SortedFileGlobalIndexReader<R
extends Closeable>
return CompletableFuture.completedFuture(Optional.empty());
}
+ @SafeVarargs
+ private final CompletableFuture<Optional<GlobalIndexResult>> complement(
+ CompletableFuture<Optional<GlobalIndexResult>>... excludeFutures) {
+ return CompletableFuture.allOf(excludeFutures)
+ .thenApply(
+ ignored -> {
+ GlobalIndexResult result =
+ totalRowCount == 0
+ ? GlobalIndexResult.createEmpty()
+ : GlobalIndexResult.fromRange(
+ new Range(0, totalRowCount
- 1));
+ for
(CompletableFuture<Optional<GlobalIndexResult>> future :
+ excludeFutures) {
+ Optional<GlobalIndexResult> excluded =
future.join();
+ if (!excluded.isPresent()) {
+ return Optional.empty();
+ }
+ result = result.andNot(excluded.get());
+ }
+ return Optional.of(result);
+ });
+ }
+
private CompletableFuture<Optional<GlobalIndexResult>> visitParallel(
Supplier<Optional<List<GlobalIndexIOMeta>>> selector,
Function<R, Optional<GlobalIndexResult>> visitor) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
index 6a97d70bcc..e2a5c978f6 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java
@@ -70,8 +70,9 @@ public class BitmapGlobalIndexer implements GlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
return new LazyFilteredBitmapReader(
- fileReader, files, keySerializer, fallbackScanMaxSize,
executor);
+ fileReader, files, keySerializer, fallbackScanMaxSize,
totalRowCount, executor);
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java
index 69c083a9fa..e0e25e79cf 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java
@@ -42,17 +42,13 @@ public class LazyFilteredBitmapReader extends
SortedFileGlobalIndexReader<Bitmap
List<GlobalIndexIOMeta> files,
KeySerializer keySerializer,
long fallbackScanMaxSize,
+ long totalRowCount,
ExecutorService executor) {
- super(files, keySerializer, fallbackScanMaxSize, executor);
+ super(files, keySerializer, fallbackScanMaxSize, totalRowCount,
executor);
this.fileReader = fileReader;
this.keySerializer = keySerializer;
}
- @Override
- protected Optional<GlobalIndexResult> visitIsNotNull(BitmapIndexReader
reader) {
- return reader.visitIsNotNull();
- }
-
@Override
protected Optional<GlobalIndexResult> visitIsNull(BitmapIndexReader
reader) {
return reader.visitIsNull();
@@ -85,11 +81,6 @@ public class LazyFilteredBitmapReader extends
SortedFileGlobalIndexReader<Bitmap
return reader.visitGreaterOrEqual(literal);
}
- @Override
- protected Optional<GlobalIndexResult> visitNotEqual(BitmapIndexReader
reader, Object literal) {
- return reader.visitNotEqual(literal);
- }
-
@Override
protected Optional<GlobalIndexResult> visitLessOrEqual(
BitmapIndexReader reader, Object literal) {
@@ -112,12 +103,6 @@ public class LazyFilteredBitmapReader extends
SortedFileGlobalIndexReader<Bitmap
return reader.visitIn(literals);
}
- @Override
- protected Optional<GlobalIndexResult> visitNotIn(
- BitmapIndexReader reader, List<Object> literals) {
- return reader.visitNotIn(literals);
- }
-
@Override
protected Optional<GlobalIndexResult> visitBetween(
BitmapIndexReader reader, Object from, Object to) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
index 24be409605..07de8ff58c 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java
@@ -98,6 +98,7 @@ public class BTreeGlobalIndexer implements GlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
return new LazyFilteredBTreeReader(
files,
@@ -105,6 +106,7 @@ public class BTreeGlobalIndexer implements GlobalIndexer {
fileReader,
cacheManager.get(),
fallbackScanMaxSize,
+ totalRowCount,
executor);
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java
index f97d7a0735..75595f1bd1 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java
@@ -51,23 +51,14 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
GlobalIndexFileReader fileReader,
CacheManager cacheManager,
long fallbackScanMaxSize,
+ long totalRowCount,
ExecutorService executor) {
- super(files, keySerializer, fallbackScanMaxSize, executor);
+ super(files, keySerializer, fallbackScanMaxSize, totalRowCount,
executor);
this.cacheManager = cacheManager;
this.fileReader = fileReader;
this.keySerializer = keySerializer;
}
- @Override
- public boolean supportsRangeComplement() {
- return true;
- }
-
- @Override
- protected Optional<GlobalIndexResult> visitIsNotNull(BTreeIndexReader
reader) {
- return reader.visitIsNotNull();
- }
-
@Override
protected Optional<GlobalIndexResult> visitIsNull(BTreeIndexReader reader)
{
return reader.visitIsNull();
@@ -105,11 +96,6 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
return reader.visitGreaterOrEqual(literal);
}
- @Override
- protected Optional<GlobalIndexResult> visitNotEqual(BTreeIndexReader
reader, Object literal) {
- return reader.visitNotEqual(literal);
- }
-
@Override
protected Optional<GlobalIndexResult> visitLessOrEqual(
BTreeIndexReader reader, Object literal) {
@@ -132,12 +118,6 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
return reader.visitIn(literals);
}
- @Override
- protected Optional<GlobalIndexResult> visitNotIn(
- BTreeIndexReader reader, List<Object> literals) {
- return reader.visitNotIn(literals);
- }
-
@Override
protected Optional<GlobalIndexResult> visitBetween(
BTreeIndexReader reader, Object from, Object to) {
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 a480ae35af..8b7d677ed3 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
@@ -807,98 +807,45 @@ class GlobalIndexEvaluatorTest {
}
@Test
- void testOffsetRangeComplementForNegativePredicates() {
+ void testOffsetDelegatesNegativePredicates() {
FieldRef fieldRef = new FieldRef(0, "a", DataTypes.INT());
AtomicBoolean isNotNullVisited = new AtomicBoolean();
- AtomicBoolean notEqualVisited = new AtomicBoolean();
- AtomicBoolean notInVisited = new AtomicBoolean();
+ AtomicInteger notEqualVisits = new AtomicInteger();
+ AtomicInteger notInVisits = new AtomicInteger();
GlobalIndexReader delegate =
new StubGlobalIndexReader(null) {
- @Override
- public boolean supportsRangeComplement() {
- return true;
- }
-
- @Override
- public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNull(
- FieldRef fieldRef) {
- return
CompletableFuture.completedFuture(Optional.of(resultOf(2, 4)));
- }
-
- @Override
- public CompletableFuture<Optional<GlobalIndexResult>>
visitEqual(
- FieldRef fieldRef, Object literal) {
- return
CompletableFuture.completedFuture(Optional.of(resultOf(1, 3)));
- }
-
- @Override
- public CompletableFuture<Optional<GlobalIndexResult>>
visitIn(
- FieldRef fieldRef, List<Object> literals) {
- return
CompletableFuture.completedFuture(Optional.of(resultOf(0, 5)));
- }
-
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(
FieldRef fieldRef) {
isNotNullVisited.set(true);
- return
CompletableFuture.completedFuture(Optional.of(resultOf(999)));
+ return
CompletableFuture.completedFuture(Optional.of(resultOf(0, 4)));
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitNotEqual(
FieldRef fieldRef, Object literal) {
- notEqualVisited.set(true);
- return
CompletableFuture.completedFuture(Optional.of(resultOf(999)));
+ notEqualVisits.incrementAndGet();
+ return CompletableFuture.completedFuture(
+ Optional.of(literal == null ? resultOf() :
resultOf(1, 3)));
}
@Override
public CompletableFuture<Optional<GlobalIndexResult>>
visitNotIn(
FieldRef fieldRef, List<Object> literals) {
- notInVisited.set(true);
- return
CompletableFuture.completedFuture(Optional.of(resultOf(999)));
+ notInVisits.incrementAndGet();
+ return CompletableFuture.completedFuture(
+ Optional.of(literals.contains(null) ?
resultOf() : resultOf(2)));
}
};
GlobalIndexReader reader = new OffsetGlobalIndexReader(delegate, 10L,
15L);
assertBitmapContainsExactly(
- reader.visitIsNotNull(fieldRef).join().get().results(), 10L,
11L, 13L, 15L);
+ reader.visitIsNotNull(fieldRef).join().get().results(), 10L,
14L);
assertBitmapContainsExactly(
- reader.visitNotEqual(fieldRef, 5).join().get().results(), 10L,
15L);
+ reader.visitNotEqual(fieldRef, 5).join().get().results(), 11L,
13L);
assertBitmapContainsExactly(
- reader.visitNotIn(fieldRef, Arrays.asList(5,
6)).join().get().results(), 11L, 13L);
- assertThat(isNotNullVisited).isFalse();
- assertThat(notEqualVisited).isFalse();
- assertThat(notInVisited).isFalse();
- }
-
- @Test
- void testOffsetRangeComplementNullAndUnsupportedPredicates() {
- FieldRef fieldRef = new FieldRef(0, "a", DataTypes.INT());
- GlobalIndexReader unsupportedEqual =
- new StubGlobalIndexReader(null) {
- @Override
- public boolean supportsRangeComplement() {
- return true;
- }
-
- @Override
- public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNull(
- FieldRef fieldRef) {
- return CompletableFuture.completedFuture(
- Optional.of(GlobalIndexResult.createEmpty()));
- }
-
- @Override
- public CompletableFuture<Optional<GlobalIndexResult>>
visitEqual(
- FieldRef fieldRef, Object literal) {
- return
CompletableFuture.completedFuture(Optional.empty());
- }
- };
-
- GlobalIndexReader reader = new
OffsetGlobalIndexReader(unsupportedEqual, 10L, 15L);
-
- assertThat(reader.visitNotEqual(fieldRef, 5).join()).isEmpty();
+ reader.visitNotIn(fieldRef, Arrays.asList(5,
6)).join().get().results(), 12L);
assertThat(reader.visitNotEqual(fieldRef,
null).join().get().results().isEmpty()).isTrue();
assertThat(
reader.visitNotIn(fieldRef, Arrays.asList(5, null))
@@ -907,6 +854,9 @@ class GlobalIndexEvaluatorTest {
.results()
.isEmpty())
.isTrue();
+ assertThat(isNotNullVisited).isTrue();
+ assertThat(notEqualVisits).hasValue(2);
+ assertThat(notInVisits).hasValue(2);
}
@Test
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
index bf0b36f667..a032a312ee 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java
@@ -110,7 +110,10 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
globalIndexer.createReader(
- fileReader, Collections.singletonList(meta),
newDirectExecutorService())) {
+ fileReader,
+ Collections.singletonList(meta),
+ 6,
+ newDirectExecutorService())) {
assertRows(reader.visitEqual(fieldRef, str("A")).join(), 0L, 5L);
assertRows(reader.visitEqual(fieldRef, null).join());
assertRows(reader.visitEqual(fieldRef, str("missing")).join());
@@ -137,6 +140,44 @@ public class LazyFilteredBitmapIndexReaderTest {
}
}
+ @Test
+ public void testTotalRowCountComplementAvoidsOpeningAllBitmapFiles()
throws Exception {
+ List<GlobalIndexIOMeta> written = new ArrayList<>();
+ written.add(writeData(Collections.singletonList(Pair.of(null, 0L))));
+ written.add(writeData(Collections.singletonList(Pair.of(str("A"),
1L))));
+ written.add(writeData(Collections.singletonList(Pair.of(str("M"),
2L))));
+ written.add(writeData(Collections.singletonList(Pair.of(str("Z"),
3L))));
+
+ CountingGlobalIndexFileReader countingReader = new
CountingGlobalIndexFileReader();
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(
+ countingReader, written, 4,
newDirectExecutorService())) {
+ assertRows(reader.visitNotEqual(fieldRef, null).join());
+ assertRows(reader.visitNotIn(fieldRef, Arrays.asList(str("M"),
null)).join());
+ assertThat(countingReader.openCount()).isZero();
+
+ assertRows(reader.visitNotEqual(fieldRef, str("M")).join(), 1L,
3L);
+ assertThat(countingReader.openCount()).isEqualTo(2);
+
+ assertRows(reader.visitNotIn(fieldRef, Arrays.asList(str("M"),
str("Z"))).join(), 1L);
+ assertThat(countingReader.openCount()).isEqualTo(3);
+
+ assertRows(reader.visitIsNotNull(fieldRef).join(), 1L, 2L, 3L);
+ assertThat(countingReader.openCount()).isEqualTo(3);
+ }
+ }
+
+ @Test
+ public void testComplementsEmptyRowDomain() throws Exception {
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(
+ fileReader, Collections.emptyList(), 0,
newDirectExecutorService())) {
+ assertRows(reader.visitIsNotNull(fieldRef).join());
+ assertRows(reader.visitNotEqual(fieldRef, str("A")).join());
+ assertRows(reader.visitNotIn(fieldRef,
Collections.singletonList(str("A"))).join());
+ }
+ }
+
@Test
public void testFlushesCompletedBitmapBeforeFinish() throws Exception {
AtomicReference<ByteArrayPositionOutputStream> output = new
AtomicReference<>();
@@ -222,7 +263,10 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
intIndexer.createReader(
- fileReader, Collections.singletonList(meta),
newDirectExecutorService())) {
+ fileReader,
+ Collections.singletonList(meta),
+ 2,
+ newDirectExecutorService())) {
assertRows(reader.visitEqual(intFieldRef, 0).join(), 1L);
}
}
@@ -240,7 +284,10 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
globalIndexer.createReader(
- fileReader, Collections.singletonList(meta),
newDirectExecutorService())) {
+ fileReader,
+ Collections.singletonList(meta),
+ 5,
+ newDirectExecutorService())) {
assertRows(reader.visitEndsWith(fieldRef, str("ta")).join(), 1L,
3L);
assertRows(reader.visitContains(fieldRef, str("ph")).join(), 0L,
2L);
assertRows(reader.visitLike(fieldRef, str("%ha%")).join(), 0L, 2L);
@@ -283,6 +330,7 @@ public class LazyFilteredBitmapIndexReaderTest {
globalIndexer.createReader(
fileReader,
Collections.singletonList(compressed),
+ 300,
newDirectExecutorService())) {
assertRows(reader.visitEqual(fieldRef, str(prefix +
"00123")).join(), 123L);
assertRows(
@@ -313,7 +361,10 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
globalIndexer.createReader(
- fileReader, Collections.singletonList(meta),
newDirectExecutorService())) {
+ fileReader,
+ Collections.singletonList(meta),
+ 2,
+ newDirectExecutorService())) {
assertThat(reader.visitEndsWith(fieldRef,
str("ta")).join()).isEmpty();
assertThat(reader.visitContains(fieldRef,
str("ph")).join()).isEmpty();
assertThat(reader.visitLike(fieldRef,
str("%ha%")).join()).isEmpty();
@@ -337,7 +388,7 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
globalIndexer.createReader(
- fileReader, Arrays.asList(first, second),
newDirectExecutorService())) {
+ fileReader, Arrays.asList(first, second), 6,
newDirectExecutorService())) {
assertRows(reader.visitEqual(fieldRef, str("B")).join(), 1L, 3L);
assertRows(reader.visitNotEqual(fieldRef, str("A")).join(), 1L,
3L, 4L);
assertRows(
@@ -361,6 +412,7 @@ public class LazyFilteredBitmapIndexReaderTest {
globalIndexer.createReader(
countingFileReader,
Arrays.asList(first, second),
+ 4,
newDirectExecutorService())) {
assertRows(reader.visitEqual(fieldRef, str("Z")).join(), 3L);
@@ -395,7 +447,7 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
globalIndexer.createReader(
- fileReader, Arrays.asList(first, second),
newDirectExecutorService())) {
+ fileReader, Arrays.asList(first, second), 4,
newDirectExecutorService())) {
assertRows(reader.visitGreaterOrEqual(fieldRef, str("Y")).join(),
2L, 3L);
assertThat(reader.visitContains(fieldRef,
str("Z")).join()).isEmpty();
}
@@ -420,7 +472,10 @@ public class LazyFilteredBitmapIndexReaderTest {
try (GlobalIndexReader reader =
globalIndexer.createReader(
- fileReader, Collections.singletonList(meta),
newDirectExecutorService())) {
+ fileReader,
+ Collections.singletonList(meta),
+ 6,
+ newDirectExecutorService())) {
assertRows(reader.visitStartsWith(fieldRef, str("tag-")).join(),
2L, 3L);
}
}
@@ -450,6 +505,7 @@ public class LazyFilteredBitmapIndexReaderTest {
globalIndexer.createReader(
countingFileReader,
Collections.singletonList(meta),
+ 250,
newDirectExecutorService())) {
assertRows(reader.visitStartsWith(fieldRef,
str("tag-match")).join(), 100L, 101L, 102L);
@@ -478,6 +534,7 @@ public class LazyFilteredBitmapIndexReaderTest {
globalIndexer.createReader(
countingFileReader,
Collections.singletonList(meta),
+ 100,
newDirectExecutorService())) {
assertRows(reader.visitEqual(fieldRef, str("tag-050")).join(),
50L);
@@ -499,6 +556,7 @@ public class LazyFilteredBitmapIndexReaderTest {
globalIndexer.createReader(
countingFileReader,
Collections.singletonList(meta),
+ 3,
newDirectExecutorService())) {
assertRows(reader.visitIsNull(fieldRef).join(), 2L);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderTest.java
index 21a9870a79..b6e5eed591 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexReaderTest.java
@@ -51,7 +51,10 @@ public class BTreeIndexReaderTest extends
AbstractIndexReaderTest {
protected GlobalIndexReader prepareDataAndCreateReader() throws Exception {
GlobalIndexIOMeta written = writeData(data);
return globalIndexer.createReader(
- fileReader, Collections.singletonList(written),
newDirectExecutorService());
+ fileReader,
+ Collections.singletonList(written),
+ dataNum,
+ newDirectExecutorService());
}
@TestTemplate
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
index 2bf888fe87..2f22974c43 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java
@@ -130,7 +130,8 @@ public class BTreeThreadSafetyTest {
executor = Executors.newFixedThreadPool(2);
List<GlobalIndexIOMeta> metas = writeMultipleFiles();
- try (GlobalIndexReader reader = globalIndexer.createReader(fileReader,
metas, executor)) {
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(fileReader, metas, DATA_NUM,
executor)) {
FieldRef ref = new FieldRef(1, "id", new IntType());
Optional<GlobalIndexResult> result =
reader.visitIsNotNull(ref).get(10, TimeUnit.SECONDS);
@@ -144,7 +145,8 @@ public class BTreeThreadSafetyTest {
executor = Executors.newFixedThreadPool(8);
List<GlobalIndexIOMeta> metas = writeMultipleFiles();
- try (GlobalIndexReader reader = globalIndexer.createReader(fileReader,
metas, executor)) {
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(fileReader, metas, DATA_NUM,
executor)) {
FieldRef ref = new FieldRef(1, "id", new IntType());
int numThreads = 32;
CountDownLatch latch = new CountDownLatch(numThreads);
@@ -187,7 +189,8 @@ public class BTreeThreadSafetyTest {
executor = Executors.newFixedThreadPool(4);
List<GlobalIndexIOMeta> metas = writeMultipleFiles();
- try (GlobalIndexReader reader = globalIndexer.createReader(fileReader,
metas, executor)) {
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(fileReader, metas, DATA_NUM,
executor)) {
FieldRef ref = new FieldRef(1, "id", new IntType());
int numThreads = 24;
CountDownLatch latch = new CountDownLatch(numThreads);
@@ -274,7 +277,7 @@ public class BTreeThreadSafetyTest {
};
try (GlobalIndexReader reader =
- globalIndexer.createReader(countingFileReader, metas,
executor)) {
+ globalIndexer.createReader(countingFileReader, metas,
DATA_NUM, executor)) {
FieldRef ref = new FieldRef(1, "id", new IntType());
int numThreads = 32;
CountDownLatch latch = new CountDownLatch(numThreads);
@@ -326,7 +329,7 @@ public class BTreeThreadSafetyTest {
};
try (GlobalIndexReader reader =
- globalIndexer.createReader(trackingFileReader, metas,
executor)) {
+ globalIndexer.createReader(trackingFileReader, metas,
DATA_NUM, executor)) {
FieldRef ref = new FieldRef(1, "id", new IntType());
// Query for value 5 — should only need to open the first file
(keys 0-999)
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
index 1fd4c509e9..7828a60bba 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java
@@ -76,7 +76,7 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
@Override
protected GlobalIndexReader prepareDataAndCreateReader() throws Exception {
List<GlobalIndexIOMeta> written = writeData();
- return globalIndexer.createReader(fileReader, written,
newDirectExecutorService());
+ return globalIndexer.createReader(fileReader, written, dataNum,
newDirectExecutorService());
}
private List<GlobalIndexIOMeta> writeData() throws Exception {
@@ -110,7 +110,8 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
FieldRef ref = new FieldRef(1, "testField", dataType);
try (GlobalIndexReader reader =
- globalIndexer.createReader(fileReader, written,
newDirectExecutorService())) {
+ globalIndexer.createReader(
+ fileReader, written, dataNum,
newDirectExecutorService())) {
GlobalIndexResult result =
reader.visitTopN(new TopN(ref, DESCENDING, NULLS_LAST,
limit)).join().get();
assertThat(result.results().getLongCardinality()).isEqualTo(limit);
@@ -146,7 +147,8 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
Object max = data.get(dataNum - 1).getKey();
try (GlobalIndexReader reader =
- globalIndexer.createReader(fileReader, written,
newDirectExecutorService())) {
+ globalIndexer.createReader(
+ fileReader, written, dataNum,
newDirectExecutorService())) {
assertThat(reader.visitBetween(ref, min, max).join()).isEmpty();
GlobalIndexResult result = reader.visitEqual(ref,
literal).join().get();
@@ -176,7 +178,8 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
Object secondFileMin = data.get(split).getKey();
try (GlobalIndexReader reader =
- globalIndexer.createReader(fileReader, written,
newDirectExecutorService())) {
+ globalIndexer.createReader(
+ fileReader, written, dataNum,
newDirectExecutorService())) {
assertThat(reader.visitBetween(ref, min, max).join()).isEmpty();
GlobalIndexResult result = reader.visitGreaterOrEqual(ref,
secondFileMin).join().get();
@@ -202,7 +205,7 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
try (GlobalIndexReader reader =
new OffsetGlobalIndexReader(
globalIndexer.createReader(
- countingReader, written,
newDirectExecutorService()),
+ countingReader, written, 4,
newDirectExecutorService()),
1000L,
1003L)) {
GlobalIndexResult result = reader.visitNotEqual(ref,
100).join().get();
@@ -324,7 +327,7 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
// Real multi-threaded executor for the reader's internal file-level
parallelism
ExecutorService readerExecutor = Executors.newFixedThreadPool(8);
try (GlobalIndexReader reader =
- stressIndexer.createReader(fileReader, written,
readerExecutor)) {
+ stressIndexer.createReader(fileReader, written, dataNum,
readerExecutor)) {
FieldRef ref = new FieldRef(1, "testField", dataType);
int concurrency = 16;
@@ -407,7 +410,7 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
new SemaphoredDelegatingExecutor(baseExecutor, 2, false);
try (GlobalIndexReader reader =
- globalIndexer.createReader(fileReader, written,
semaphoredExecutor)) {
+ globalIndexer.createReader(fileReader, written, dataNum,
semaphoredExecutor)) {
FieldRef ref = new FieldRef(1, "testField", dataType);
Random random = new Random(42);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexer.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexer.java
index a7fdc0d49d..ac8ec32a6a 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexer.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexer.java
@@ -58,6 +58,7 @@ public class TestFullTextGlobalIndexer implements
GlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
checkArgument(files.size() == 1, "Expected exactly one index file per
shard");
return new TestFullTextGlobalIndexReader(fileReader, files.get(0));
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestMultiFieldVectorGlobalIndexer.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestMultiFieldVectorGlobalIndexer.java
index e984dd03b2..b3570b0be2 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestMultiFieldVectorGlobalIndexer.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestMultiFieldVectorGlobalIndexer.java
@@ -81,6 +81,7 @@ class TestMultiFieldVectorGlobalIndexer implements
VectorGlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
List<GlobalIndexIOMeta> vectorFiles = new ArrayList<>();
List<GlobalIndexIOMeta> scalarFiles = new ArrayList<>();
@@ -92,8 +93,8 @@ class TestMultiFieldVectorGlobalIndexer implements
VectorGlobalIndexer {
checkArgument(
scalarFiles.size() == 1, "Expected one scalar companion file,
got: %s", files);
return new MultiColumnReader(
- vectorIndexer.createReader(fileReader, vectorFiles, executor),
- scalarIndexer.createReader(fileReader, scalarFiles, executor));
+ vectorIndexer.createReader(fileReader, vectorFiles,
totalRowCount, executor),
+ scalarIndexer.createReader(fileReader, scalarFiles,
totalRowCount, executor));
}
@Override
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
index eb0365427a..a114189ebb 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
@@ -105,6 +105,7 @@ public class TestVectorGlobalIndexer implements
VectorGlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
checkArgument(files.size() == 1, "Expected exactly one index file per
shard");
return new TestVectorGlobalIndexReader(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java
index 9d6acaedde..d391743785 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java
@@ -442,7 +442,10 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
() ->
new OffsetGlobalIndexReader(
globalIndexer.createReader(
- indexFileReadWrite,
globalMetas, executor),
+ indexFileReadWrite,
+ globalMetas,
+ range.count(),
+ executor),
range.from,
range.to),
executor));
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 31f95a8839..2bdebce47d 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
@@ -113,7 +113,7 @@ public class PrimaryKeyFullTextBucketSearch {
if (include != null && include.isEmpty()) {
continue;
}
- GlobalIndexReader reader = readerFactory.create(payload);
+ GlobalIndexReader reader = readerFactory.create(payload,
totalRowCount);
CompletableFuture<Optional<ScoredGlobalIndexResult>> future;
try {
FullTextSearch predicate = new FullTextSearch(column, query,
limit);
@@ -213,7 +213,7 @@ public class PrimaryKeyFullTextBucketSearch {
/** Creates one independently closeable reader for an immutable payload
archive. */
@FunctionalInterface
public interface ReaderFactory {
- GlobalIndexReader create(IndexFileMeta payload);
+ GlobalIndexReader create(IndexFileMeta payload, long totalRowCount);
}
private static class PayloadRequest {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
index 75274336a9..1aff314898 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkvector/PkVectorAnnSegmentSearcher.java
@@ -203,6 +203,7 @@ public class PkVectorAnnSegmentSearcher {
indexer.createReader(
meta -> fileIO.newInputStream(meta.filePath()),
Collections.singletonList(ioMeta),
+ segment.rowCount(),
executor);
try {
VectorSearch search = new VectorSearch(query, limit,
vectorField.name(), searchOptions);
@@ -299,6 +300,7 @@ public class PkVectorAnnSegmentSearcher {
indexer.createReader(
meta -> fileIO.newInputStream(meta.filePath()),
Collections.singletonList(ioMeta),
+ segment.rowCount(),
executor);
try {
BatchVectorSearch search =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
index b35a1eb45c..83ea140270 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
@@ -258,7 +258,11 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
FileIO fileIO = table.fileIO();
GlobalIndexFileReader indexFileReader = m ->
fileIO.newInputStream(m.filePath());
GlobalIndexReader reader =
- globalIndexer.createReader(indexFileReader, indexIOMetaList,
executor);
+ globalIndexer.createReader(
+ indexFileReader,
+ indexIOMetaList,
+ rowRangeEnd - rowRangeStart + 1,
+ executor);
VectorSearch vectorSearch =
new VectorSearch(vector, searchLimit, vectorColumn.name(),
options)
.withIncludeRowIds(includeRowIds);
@@ -287,7 +291,11 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
FileIO fileIO = table.fileIO();
GlobalIndexFileReader indexFileReader = m ->
fileIO.newInputStream(m.filePath());
GlobalIndexReader reader =
- globalIndexer.createReader(indexFileReader, indexIOMetaList,
executor);
+ globalIndexer.createReader(
+ indexFileReader,
+ indexIOMetaList,
+ rowRangeEnd - rowRangeStart + 1,
+ executor);
BatchVectorSearch batchVectorSearch =
new BatchVectorSearch(vectors, searchLimit,
vectorColumn.name(), options)
.withIncludeRowIds(includeRowIds);
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 780155f837..40e492968c 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
@@ -254,7 +254,11 @@ public class DataEvolutionFullTextRead implements
FullTextRead {
meta.indexMeta()));
}
GlobalIndexReader reader =
- globalIndexer.createReader(indexFileReader, indexIOMetaList,
executor);
+ globalIndexer.createReader(
+ indexFileReader,
+ indexIOMetaList,
+ rowRangeEnd - rowRangeStart + 1,
+ executor);
FullTextSearch fullTextSearch =
new FullTextSearch(
textColumn.name(),
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 c514301be0..edc56a372f 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
@@ -189,7 +189,7 @@ public class PrimaryKeyFullTextRead implements FullTextRead
{
indexFileHandler.pkFullTextIndex(
split.dataSplit().partition(),
split.dataSplit().bucket());
return new PrimaryKeyFullTextBucketSearch(
- payload -> {
+ (payload, totalRowCount) -> {
GlobalIndexMeta meta =
checkNotNull(payload.globalIndexMeta());
GlobalIndexIOMeta ioMeta =
new GlobalIndexIOMeta(
@@ -198,7 +198,10 @@ public class PrimaryKeyFullTextRead implements
FullTextRead {
meta.indexMeta());
GlobalIndexReader reader =
indexer.createReader(
- archiveReader,
Collections.singletonList(ioMeta), executor);
+ archiveReader,
+ Collections.singletonList(ioMeta),
+ totalRowCount,
+ executor);
return reader;
});
}
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 adea4965ac..7bc04f5d6f 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
@@ -83,7 +83,10 @@ public final class PrimaryKeySortedIndexScan {
public interface ReaderFactory {
GlobalIndexReader create(
- FilePlan file, PrimaryKeyIndexDefinition definition,
List<IndexFileMeta> payloads);
+ FilePlan file,
+ PrimaryKeyIndexDefinition definition,
+ List<IndexFileMeta> payloads,
+ long totalRowCount);
}
static ReaderFactory readerFactory(
@@ -92,7 +95,7 @@ public final class PrimaryKeySortedIndexScan {
ExecutorService executor =
GlobalIndexReadThreadPool.getExecutorService(options.get(GLOBAL_INDEX_THREAD_NUM));
GlobalIndexFileReader fileReader = meta ->
fileIO.newInputStream(meta.filePath());
- return (file, definition, payloads) -> {
+ return (file, definition, payloads, totalRowCount) -> {
IndexPathFactory indexPathFactory =
pathFactories.get(file.sourceSplit().partition(),
file.sourceSplit().bucket());
List<GlobalIndexIOMeta> ioMetas = new ArrayList<>(payloads.size());
@@ -109,7 +112,7 @@ public final class PrimaryKeySortedIndexScan {
definition.indexType(),
rowType.getField(definition.fieldId()),
definition.options());
- return indexer.createReader(fileReader, ioMetas, executor);
+ return indexer.createReader(fileReader, ioMetas, totalRowCount,
executor);
};
}
@@ -257,15 +260,19 @@ public final class PrimaryKeySortedIndexScan {
}
SharedGlobalIndexReader reader =
sharedReaders.get(group.get());
if (reader == null) {
+ PkSortedIndexGroup indexGroup =
group.get();
+ long totalRowCount =
+
totalRowCount(indexGroup.sourceFiles());
reader =
new SharedGlobalIndexReader(
-
group.get().sourceFiles(),
+
indexGroup.sourceFiles(),
() ->
readerFactory.create(
file,
definition,
-
group.get().payloads()));
- sharedReaders.put(group.get(), reader);
+
indexGroup.payloads(),
+
totalRowCount));
+ sharedReaders.put(indexGroup, reader);
}
return Collections.singletonList(
fileLocalReader(file, group.get(),
reader));
@@ -312,6 +319,14 @@ public final class PrimaryKeySortedIndexScan {
return new FileLocalGlobalIndexReader(reader, sourceIndex);
}
+ private static long totalRowCount(List<PrimaryKeyIndexSourceFile>
sourceFiles) {
+ long totalRowCount = 0;
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
+ totalRowCount = Math.addExact(totalRowCount,
sourceFile.rowCount());
+ }
+ return totalRowCount;
+ }
+
private static void rethrowIfInterrupted(RuntimeException exception) {
if (Thread.currentThread().isInterrupted()) {
throw exception;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
index ffbfab5ff6..59583e45e1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PkFullTextIndexFileTest.java
@@ -273,6 +273,7 @@ class PkFullTextIndexFileTest {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
throw new UnsupportedOperationException();
}
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 d5e148341e..231343cac5 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
@@ -71,7 +71,7 @@ class PrimaryKeyFullTextBucketSearchTest {
PrimaryKeyFullTextBucketSearch search =
new PrimaryKeyFullTextBucketSearch(
- payload -> {
+ (payload, ignoredTotalRowCount) -> {
String source =
PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
.sourceFile()
@@ -101,7 +101,7 @@ class PrimaryKeyFullTextBucketSearchTest {
AtomicInteger closes = new AtomicInteger();
PrimaryKeyFullTextBucketSearch search =
new PrimaryKeyFullTextBucketSearch(
- payload -> reader(scores(0, 1F, 2, 10F), closes));
+ (payload, ignoredTotalRowCount) -> reader(scores(0,
1F, 2, 10F), closes));
PrimaryKeyFullTextSearchSplit split =
new PrimaryKeyFullTextSearchSplit(
dataSplit(Collections.singletonList(dataFile("a"))),
@@ -125,7 +125,10 @@ class PrimaryKeyFullTextBucketSearchTest {
AtomicInteger closes = new AtomicInteger();
PrimaryKeyFullTextBucketSearch search =
new PrimaryKeyFullTextBucketSearch(
- ignored -> reader(scores(0, 10F, 3, 9F, 4, 8F, 5, 7F),
closes));
+ (ignored, totalRowCount) -> {
+ assertThat(totalRowCount).isEqualTo(6);
+ return reader(scores(0, 10F, 3, 9F, 4, 8F, 5, 7F),
closes);
+ });
PrimaryKeyFullTextSearchSplit split =
new PrimaryKeyFullTextSearchSplit(
dataSplit(Arrays.asList(dataFile("a"), dataFile("b"))),
@@ -158,7 +161,7 @@ class PrimaryKeyFullTextBucketSearchTest {
AtomicInteger closes = new AtomicInteger();
PrimaryKeyFullTextBucketSearch search =
new PrimaryKeyFullTextBucketSearch(
- ignored -> reader(Collections.emptyMap(), closes));
+ (ignored, ignoredTotalRowCount) ->
reader(Collections.emptyMap(), closes));
PrimaryKeyFullTextSearchSplit split =
new PrimaryKeyFullTextSearchSplit(
dataSplit(Collections.singletonList(dataFile("large",
rowCount))),
@@ -182,7 +185,7 @@ class PrimaryKeyFullTextBucketSearchTest {
AtomicInteger closes = new AtomicInteger();
PrimaryKeyFullTextBucketSearch search =
new PrimaryKeyFullTextBucketSearch(
- payload ->
+ (payload, ignoredTotalRowCount) ->
new FullTextOnlyReader() {
@Override
public
CompletableFuture<Optional<ScoredGlobalIndexResult>>
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
index db2d6d0b94..a57363e4b1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pksorted/PkSortedIndexBuilderTest.java
@@ -238,6 +238,7 @@ class PkSortedIndexBuilderTest {
.createReader(
new GlobalIndexFileReadWrite(fileIO,
pathFactory),
ioMetas,
+ payload.rowCount(),
executor)) {
FieldRef fieldRef = new FieldRef(7, "indexed", DataTypes.INT());
GlobalIndexResult result =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
index f4d8243a3c..bdb286c508 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexBatchScanTest.java
@@ -223,7 +223,8 @@ class PrimaryKeySortedIndexBatchScanTest {
table,
snapshotReader,
mock(TableQueryAuth.class),
- (ignoredFile, ignoredDefinition, ignoredPayloads) ->
reader);
+ (ignoredFile, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) ->
+ reader);
scan.withFilter(predicate);
return new ScanFixture(dataFile, scan);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
index 34429e545c..0aa9e0ebc9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeySortedIndexResultTest.java
@@ -95,7 +95,7 @@ class PrimaryKeySortedIndexResultTest {
rowType,
predicate,
Collections.singletonList(definition),
- (file, ignoredDefinition, ignoredPayloads) -> {
+ (file, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) -> {
if (file.dataFile().fileName().equals("indexed")) {
return reader(1, 2, 4);
} else if
(file.dataFile().fileName().equals("empty")) {
@@ -152,7 +152,7 @@ class PrimaryKeySortedIndexResultTest {
rowType,
predicate,
Collections.singletonList(definition),
- (file, ignoredDefinition, ignoredPayloads) ->
+ (file, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) ->
file.dataFile().fileName().equals("first") ?
reader(1) : reader(2));
PrimaryKeySortedIndexResult result = new
PrimaryKeySortedIndexResult(evaluated);
@@ -195,7 +195,8 @@ class PrimaryKeySortedIndexResultTest {
rowType,
predicate,
Collections.singletonList(definition),
- (ignoredFile, ignoredDefinition, ignoredPayloads) ->
reader(positions));
+ (ignoredFile, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) ->
+ reader(positions));
PrimaryKeySortedIndexResult result = new
PrimaryKeySortedIndexResult(evaluated);
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 af44371136..c704589af7 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
@@ -191,7 +191,7 @@ class PrimaryKeySortedIndexScanTest {
rowType,
predicate,
Collections.singletonList(definition),
- (ignoredFile, ignoredDefinition, payloads) -> {
+ (ignoredFile, ignoredDefinition, payloads,
ignoredTotalRowCount) -> {
readersCreated.incrementAndGet();
return reader;
});
@@ -248,9 +248,10 @@ class PrimaryKeySortedIndexScanTest {
rowType,
predicate,
Collections.singletonList(definition),
- (ignoredFile, ignoredDefinition, payloads) -> {
+ (ignoredFile, ignoredDefinition, payloads,
totalRowCount) -> {
readersCreated.incrementAndGet();
assertThat(payloads).containsExactly(mergedPayload);
+ assertThat(totalRowCount).isEqualTo(5);
return reader;
});
PrimaryKeySortedIndexResult result = new
PrimaryKeySortedIndexResult(evaluated);
@@ -297,14 +298,16 @@ class PrimaryKeySortedIndexScanTest {
rowType,
PredicateBuilder.and(builder.equal(0, 42),
builder.equal(1, 99)),
Collections.singletonList(definition),
- (ignoredFile, ignoredDefinition, ignoredPayloads) ->
reader);
+ (ignoredFile, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) ->
+ reader);
PrimaryKeySortedIndexScan.EvaluatedPlan orResult =
PrimaryKeySortedIndexScan.evaluate(
plan,
rowType,
PredicateBuilder.or(builder.equal(0, 42),
builder.equal(1, 99)),
Collections.singletonList(definition),
- (ignoredFile, ignoredDefinition, ignoredPayloads) ->
reader);
+ (ignoredFile, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) ->
+ reader);
assertThat(andResult.files().get(0).result()).isPresent();
assertThat(andResult.files().get(0).result().get().results()).containsExactly(2L);
@@ -346,7 +349,8 @@ class PrimaryKeySortedIndexScanTest {
rowType,
predicate,
Collections.singletonList(definition),
- (file, ignoredDefinition, ignoredPayloads) ->
failedReader);
+ (file, ignoredDefinition, ignoredPayloads,
ignoredTotalRowCount) ->
+ failedReader);
assertThat(evaluated.files()).hasSize(2);
assertThat(evaluated.files().get(0).result()).isEmpty();
diff --git
a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java
b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java
index 0167e24066..3ae8620624 100644
---
a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java
+++
b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexer.java
@@ -61,6 +61,7 @@ public class ESIndexGlobalIndexer implements
VectorGlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
ESIndexGlobalIndexReader reader =
new ESIndexGlobalIndexReader(fileReader, files, fields,
indexOptions, executor);
diff --git
a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java
b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java
index 952ea8b093..c693225c3a 100644
---
a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java
+++
b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerExecutorTest.java
@@ -66,7 +66,7 @@ class ESIndexGlobalIndexerExecutorTest {
ExecutorService caller = Executors.newSingleThreadExecutor();
try {
ESIndexGlobalIndexer indexer = new ESIndexGlobalIndexer(FIELDS,
new Options());
- GlobalIndexReader reader = indexer.createReader(meta -> null,
oneFile(), caller);
+ GlobalIndexReader reader = indexer.createReader(meta -> null,
oneFile(), 1, caller);
assertSame(
caller,
((ESIndexGlobalIndexReader) reader).queryExecutor(),
@@ -80,7 +80,7 @@ class ESIndexGlobalIndexerExecutorTest {
void inlineValidationFailuresCompleteTheReturnedFutureExceptionally()
throws Exception {
ESIndexGlobalIndexer indexer = new ESIndexGlobalIndexer(FIELDS, new
Options());
ESIndexGlobalIndexReader reader =
- (ESIndexGlobalIndexReader) indexer.createReader(meta -> null,
oneFile(), null);
+ (ESIndexGlobalIndexReader) indexer.createReader(meta -> null,
oneFile(), 1, null);
try {
CompletableFuture<?> future =
reader.visitFullTextSearch(new FullTextSearch("k",
"{not-json", 1));
diff --git
a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java
b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java
index fceab62411..505de4d53e 100644
---
a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java
+++
b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexVectorMetricTest.java
@@ -68,6 +68,7 @@ class ESIndexVectorMetricTest {
indexer.createReader(
meta -> null,
List.of(vectorFile(tempDir, "persisted-l2",
"euclidean")),
+ 1,
null);
try {
assertThat(((ESIndexGlobalIndexReader)
reader).primaryVectorMetric()).isEqualTo("l2");
@@ -82,13 +83,14 @@ class ESIndexVectorMetricTest {
ESIndexGlobalIndexer indexer = indexer("cosine");
GlobalIndexReader first =
indexer.createReader(
- meta -> null, List.of(vectorFile(tempDir, "first",
"euclidean")), null);
+ meta -> null, List.of(vectorFile(tempDir, "first",
"euclidean")), 1, null);
try {
assertThatThrownBy(
() ->
indexer.createReader(
meta -> null,
List.of(vectorFile(tempDir,
"second", "cosine")),
+ 1,
null))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("different vector metrics")
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorReadTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorReadTest.java
index 9bd45a7776..a39e777168 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorReadTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorReadTest.java
@@ -543,6 +543,7 @@ public class FlinkDataEvolutionVectorReadTest {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
throw new UnsupportedOperationException();
}
diff --git
a/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexer.java
b/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexer.java
index 564cafa809..c8c0bb3867 100644
---
a/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexer.java
+++
b/paimon-full-text/src/main/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexer.java
@@ -50,6 +50,7 @@ public class NativeFullTextGlobalIndexer implements
GlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
return new NativeFullTextGlobalIndexReader(fileReader, files,
executor);
}
diff --git
a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexTest.java
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexTest.java
index a129aec198..e5d8f482ca 100644
---
a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexTest.java
+++
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativeFullTextGlobalIndexTest.java
@@ -402,7 +402,7 @@ public class NativeFullTextGlobalIndexTest {
try (NativeFullTextGlobalIndexReader reader =
(NativeFullTextGlobalIndexReader)
- indexer.createReader(fileReader, metas,
newDirectExecutorService())) {
+ indexer.createReader(fileReader, metas, 1,
newDirectExecutorService())) {
FullTextSearch search = new FullTextSearch("text",
matchQuery("indexer"), 10);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
diff --git
a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
index ce1100f590..a9ca34dc42 100644
---
a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
+++
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
@@ -131,10 +131,11 @@ class NativePrimaryKeyFullTextIndexTest {
GlobalIndexer indexer = GlobalIndexer.create("full-text", TEXT_FIELD,
options);
PrimaryKeyFullTextBucketSearch search =
new PrimaryKeyFullTextBucketSearch(
- payload ->
+ (payload, totalRowCount) ->
indexer.createReader(
fileReader(),
Collections.singletonList(toIOMeta(payload)),
+ totalRowCount,
newDirectExecutorService()));
List<List<PrimaryKeySearchPosition>> rankings =
diff --git
a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
index 69b73a6bce..87ca7d4b47 100644
---
a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
+++
b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
@@ -50,6 +50,7 @@ public class LuminaVectorGlobalIndexer implements
VectorGlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
return new LuminaVectorGlobalIndexReader(fileReader, files, fieldType,
options, executor);
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
index 2479118c32..dd13cd365b 100644
---
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorReadTest.java
@@ -440,6 +440,7 @@ public class SparkDataEvolutionVectorReadTest {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
throw new UnsupportedOperationException();
}
diff --git
a/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
b/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
index 37e20e0af5..2498e8d31b 100644
---
a/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
+++
b/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
@@ -79,6 +79,7 @@ public class NativeVectorGlobalIndexer implements
VectorGlobalIndexer {
public GlobalIndexReader createReader(
GlobalIndexFileReader fileReader,
List<GlobalIndexIOMeta> files,
+ long totalRowCount,
ExecutorService executor) {
return new NativeVectorGlobalIndexReader(fileReader, files, fieldType,
executor);
}
diff --git
a/paimon-vector/src/test/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexTest.java
b/paimon-vector/src/test/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexTest.java
index 4995f3e886..eb7750915a 100644
---
a/paimon-vector/src/test/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexTest.java
+++
b/paimon-vector/src/test/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexTest.java
@@ -508,7 +508,8 @@ public class NativeVectorGlobalIndexTest {
GlobalIndexFileReader fileReader = createFileReader(indexPath);
try (NativeVectorGlobalIndexReader reader =
- (NativeVectorGlobalIndexReader)
indexer.createReader(fileReader, metas, executor)) {
+ (NativeVectorGlobalIndexReader)
+ indexer.createReader(fileReader, metas,
vectors.length, executor)) {
VectorSearch vectorSearch = new VectorSearch(vectors[0], 2,
fieldName);
ScoredGlobalIndexResult result =
reader.visitVectorSearch(vectorSearch).join().get();
assertThat(result.results().getLongCardinality()).isEqualTo(2);
