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 d67113eb7b [core] Support TopN pushdown for BTree global indexes
(#8991)
d67113eb7b is described below
commit d67113eb7b4338d3404de4132ec922c9357341b1
Author: umi <[email protected]>
AuthorDate: Tue Aug 4 19:59:40 2026 +0800
[core] Support TopN pushdown for BTree global indexes (#8991)
---
.../paimon/globalindex/GlobalIndexEvaluator.java | 23 +-
.../paimon/globalindex/GlobalIndexReader.java | 11 +
.../globalindex/OffsetGlobalIndexReader.java | 6 +
.../globalindex/SortedFileGlobalIndexReader.java | 7 +
.../paimon/globalindex/TopNGlobalIndexResult.java | 205 ++++++++++++++++
.../paimon/globalindex/UnionGlobalIndexReader.java | 6 +
.../paimon/globalindex/btree/BTreeIndexReader.java | 107 ++++++++-
.../globalindex/btree/LazyFilteredBTreeReader.java | 7 +
.../java/org/apache/paimon/sst/BlockReader.java | 4 +
.../apache/paimon/sst/ReverseBlockIterator.java | 64 +++++
.../java/org/apache/paimon/sst/SstFileReader.java | 36 +++
.../globalindex/GlobalIndexEvaluatorTest.java | 43 ++++
.../globalindex/TopNGlobalIndexResultTest.java | 169 +++++++++++++
.../globalindex/btree/BTreeIndexReaderTest.java | 103 ++++++++
.../btree/LazyFilteredBTreeIndexReaderTest.java | 38 ++-
.../org/apache/paimon/sst/BlockIteratorTest.java | 15 +-
.../globalindex/BTreeTopNIndexFileSelector.java | 164 +++++++++++++
.../paimon/globalindex/DataEvolutionBatchScan.java | 85 ++++++-
.../DataEvolutionGlobalIndexScanner.java | 111 ++++++++-
.../BTreeTopNIndexFileSelectorTest.java | 237 +++++++++++++++++++
.../paimon/table/BtreeGlobalIndexTableTest.java | 262 +++++++++++++++++++++
.../paimon/spark/sql/RowTrackingTestBase.scala | 30 +++
22 files changed, 1724 insertions(+), 9 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 0f8a0e3ccd..49958cc3be 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
@@ -29,6 +29,7 @@ import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.LeafTernaryFunction;
import org.apache.paimon.predicate.Or;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IOUtils;
@@ -50,6 +51,8 @@ import java.util.concurrent.ExecutionException;
import java.util.function.IntFunction;
import java.util.stream.Collectors;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
/** Predicate for filtering data using global indexes. */
public class GlobalIndexEvaluator implements Closeable {
@@ -68,8 +71,26 @@ public class GlobalIndexEvaluator implements Closeable {
if (predicate == null) {
return Optional.empty();
}
+ return awaitGlobalIndexResult(visitAsync(predicate));
+ }
+
+ public Optional<GlobalIndexResult> evaluateTopN(TopN topN) {
+ FieldRef fieldRef = topN.orders().get(0).field();
+ int fieldId = rowType.getField(fieldRef.name()).id();
+ Collection<GlobalIndexReader> readers =
+ indexReadersCache.computeIfAbsent(fieldId,
readersFunction::apply);
+
+ if (readers.isEmpty()) {
+ return Optional.empty();
+ }
+ checkArgument(readers.size() == 1, "TopN expects one aggregated global
index reader.");
+ return
awaitGlobalIndexResult(readers.iterator().next().visitTopN(topN));
+ }
+
+ private Optional<GlobalIndexResult> awaitGlobalIndexResult(
+ CompletableFuture<Optional<GlobalIndexResult>> future) {
try {
- return visitAsync(predicate).get();
+ return future.get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Interrupted during index evaluation",
e);
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 807ecb8aa0..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
@@ -23,6 +23,7 @@ import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FullTextSearch;
import org.apache.paimon.predicate.FunctionVisitor;
import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.predicate.VectorSearch;
import java.io.Closeable;
@@ -80,6 +81,16 @@ public interface GlobalIndexReader
throw new UnsupportedOperationException();
}
+ /**
+ * Returns row candidates for the given TopN predicate.
+ *
+ * <p>The result may contain more than {@link TopN#limit()} rows when this
reader owns multiple
+ * independent index files. Callers must still apply the final TopN
operation.
+ */
+ default CompletableFuture<Optional<GlobalIndexResult>> visitTopN(TopN
topN) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
/** Batch search; result {@code i} matches vector {@code i}. */
default CompletableFuture<List<Optional<ScoredGlobalIndexResult>>>
visitBatchVectorSearch(
BatchVectorSearch batchVectorSearch) {
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 7b513a2c39..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
@@ -21,6 +21,7 @@ package org.apache.paimon.globalindex;
import org.apache.paimon.predicate.BatchVectorSearch;
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 java.io.IOException;
@@ -168,6 +169,11 @@ public class OffsetGlobalIndexReader implements
GlobalIndexReader {
});
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitTopN(TopN topN)
{
+ return wrapped.visitTopN(topN).thenApply(this::applyOffset);
+ }
+
private Optional<GlobalIndexResult>
applyOffset(Optional<GlobalIndexResult> result) {
return result.map(r -> r.offset(offset));
}
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 5d0195b3fc..f16c2036fe 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
@@ -49,6 +49,7 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
implements GlobalIndexReader {
private final SortedFileMetaSelector fileSelector;
+ private final List<GlobalIndexIOMeta> files;
private final long fallbackScanMaxSize;
private final Map<Path, R> readerCache;
private final ExecutorService executor;
@@ -59,6 +60,7 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
long fallbackScanMaxSize,
ExecutorService executor) {
this.fileSelector = new SortedFileMetaSelector(files, keySerializer);
+ this.files = new ArrayList<>(files);
this.fallbackScanMaxSize = fallbackScanMaxSize;
this.readerCache = new ConcurrentHashMap<>();
this.executor = executor;
@@ -391,6 +393,11 @@ public abstract class SortedFileGlobalIndexReader<R
extends Closeable>
.thenApply(v -> unionResults(futures));
}
+ protected CompletableFuture<Optional<GlobalIndexResult>> visitAllFiles(
+ Function<R, Optional<GlobalIndexResult>> visitor) {
+ return visitSelectedFiles(Optional.of(files), visitor);
+ }
+
private R getOrCreateReader(GlobalIndexIOMeta meta) {
return readerCache.computeIfAbsent(meta.filePath(), ignored ->
openReader(meta));
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/TopNGlobalIndexResult.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/TopNGlobalIndexResult.java
new file mode 100644
index 0000000000..6d4e05b9fb
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/TopNGlobalIndexResult.java
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.globalindex.btree.BTreeIndexReader.KeyRowIds;
+import org.apache.paimon.predicate.SortValue;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+/**
+ * A bounded global index result which retains sort keys while merging TopN
candidates.
+ *
+ * <p>Key groups are ordered by the requested sort direction and their row ids
are ordered
+ * ascending. Merging two compatible results combines equal keys and keeps
only the globally best
+ * {@code limit} row ids. Merging with a plain bitmap result falls back to a
conservative bitmap
+ * union because the other result has no sort keys.
+ */
+public final class TopNGlobalIndexResult implements GlobalIndexResult {
+
+ private final List<KeyRowIds> keyRowIds;
+ private final Comparator<Object> keyComparator;
+ private final SortValue.SortDirection direction;
+ private final SortValue.NullOrdering nullOrdering;
+ private final int limit;
+ private final RoaringNavigableMap64 results;
+
+ private TopNGlobalIndexResult(
+ List<KeyRowIds> keyRowIds,
+ Comparator<Object> keyComparator,
+ SortValue.SortDirection direction,
+ SortValue.NullOrdering nullOrdering,
+ int limit) {
+ this.keyComparator = keyComparator;
+ this.direction = direction;
+ this.nullOrdering = nullOrdering;
+ this.limit = limit;
+
+ List<KeyRowIds> sorted = new ArrayList<>(keyRowIds);
+ sorted.sort(keyRowIdsComparator());
+ this.keyRowIds = Collections.unmodifiableList(mergeAndLimit(sorted));
+ this.results = toBitmap(this.keyRowIds);
+ }
+
+ public static TopNGlobalIndexResult create(
+ List<KeyRowIds> keyRowIds,
+ Comparator<Object> keyComparator,
+ SortValue.NullOrdering nullOrdering,
+ int limit) {
+ return create(
+ keyRowIds, keyComparator, SortValue.SortDirection.DESCENDING,
nullOrdering, limit);
+ }
+
+ public static TopNGlobalIndexResult create(
+ List<KeyRowIds> keyRowIds,
+ Comparator<Object> keyComparator,
+ SortValue.SortDirection direction,
+ SortValue.NullOrdering nullOrdering,
+ int limit) {
+ Preconditions.checkArgument(limit >= 0, "TopN limit must not be
negative.");
+ return new TopNGlobalIndexResult(keyRowIds, keyComparator, direction,
nullOrdering, limit);
+ }
+
+ @Override
+ public RoaringNavigableMap64 results() {
+ return results;
+ }
+
+ @Override
+ public TopNGlobalIndexResult offset(long startOffset) {
+ if (startOffset == 0) {
+ return this;
+ }
+
+ List<KeyRowIds> offsetKeyRowIds = new ArrayList<>(keyRowIds.size());
+ for (KeyRowIds keyRowIds : keyRowIds) {
+ long[] rowIds = keyRowIds.rowIds();
+ long[] offsetRowIds = new long[rowIds.length];
+ for (int i = 0; i < rowIds.length; i++) {
+ offsetRowIds[i] = rowIds[i] + startOffset;
+ }
+ offsetKeyRowIds.add(new KeyRowIds(keyRowIds.key(), offsetRowIds));
+ }
+ return create(offsetKeyRowIds, keyComparator, direction, nullOrdering,
limit);
+ }
+
+ @Override
+ public GlobalIndexResult or(GlobalIndexResult other) {
+ if (!(other instanceof TopNGlobalIndexResult)) {
+ if (other.results().isEmpty()) {
+ return this;
+ }
+ return GlobalIndexResult.super.or(other);
+ }
+
+ TopNGlobalIndexResult sortedOther = (TopNGlobalIndexResult) other;
+ Preconditions.checkArgument(
+ limit == sortedOther.limit,
+ "Cannot merge sorted global index results with different TopN
limits.");
+ Preconditions.checkArgument(
+ direction == sortedOther.direction,
+ "Cannot merge sorted global index results with different sort
directions.");
+ Preconditions.checkArgument(
+ nullOrdering == sortedOther.nullOrdering,
+ "Cannot merge sorted global index results with different null
ordering.");
+
+ if (sortedOther.keyRowIds.isEmpty()) {
+ return this;
+ }
+ if (keyRowIds.isEmpty()) {
+ return sortedOther;
+ }
+
+ List<KeyRowIds> merged = new ArrayList<>(keyRowIds.size() +
sortedOther.keyRowIds.size());
+ merged.addAll(keyRowIds);
+ merged.addAll(sortedOther.keyRowIds);
+ return new TopNGlobalIndexResult(merged, keyComparator, direction,
nullOrdering, limit);
+ }
+
+ private List<KeyRowIds> mergeAndLimit(List<KeyRowIds> sorted) {
+ List<KeyRowIds> result = new ArrayList<>(Math.min(limit,
sorted.size()));
+ int remaining = limit;
+ int position = 0;
+ while (remaining > 0 && position < sorted.size()) {
+ Object key = sorted.get(position).key();
+ RoaringNavigableMap64 sameKeyRowIds = new RoaringNavigableMap64();
+ do {
+ for (long rowId : sorted.get(position).rowIds()) {
+ sameKeyRowIds.add(rowId);
+ }
+ position++;
+ } while (position < sorted.size() && compareKeys(key,
sorted.get(position).key()) == 0);
+
+ int count = (int) Math.min((long) remaining,
sameKeyRowIds.getLongCardinality());
+ if (count > 0) {
+ long[] limitedRowIds = new long[count];
+ int index = 0;
+ for (long rowId : sameKeyRowIds) {
+ limitedRowIds[index++] = rowId;
+ if (index == count) {
+ break;
+ }
+ }
+ result.add(new KeyRowIds(key, limitedRowIds));
+ remaining -= count;
+ }
+ }
+ return result;
+ }
+
+ private Comparator<KeyRowIds> keyRowIdsComparator() {
+ return (left, right) -> compareKeys(left.key(), right.key());
+ }
+
+ private int compareKeys(@Nullable Object left, @Nullable Object right) {
+ if (left == null && right == null) {
+ return 0;
+ }
+ if (left == null) {
+ return nullOrdering == SortValue.NullOrdering.NULLS_FIRST ? -1 : 1;
+ }
+ if (right == null) {
+ return nullOrdering == SortValue.NullOrdering.NULLS_FIRST ? 1 : -1;
+ }
+ return direction == SortValue.SortDirection.ASCENDING
+ ? keyComparator.compare(left, right)
+ : keyComparator.compare(right, left);
+ }
+
+ private static RoaringNavigableMap64 toBitmap(List<KeyRowIds> keyRowIds) {
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ for (KeyRowIds group : keyRowIds) {
+ for (long rowId : group.rowIds()) {
+ bitmap.add(rowId);
+ }
+ }
+ return bitmap;
+ }
+
+ List<KeyRowIds> keyRowIds() {
+ return keyRowIds;
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
index 473eb24d0a..0e249b5df7 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java
@@ -19,6 +19,7 @@
package org.apache.paimon.globalindex;
import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.predicate.VectorSearch;
import org.apache.paimon.utils.IOUtils;
@@ -174,6 +175,11 @@ public class UnionGlobalIndexReader implements
GlobalIndexReader {
});
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitTopN(TopN topN)
{
+ return unionAsync(reader -> reader.visitTopN(topN));
+ }
+
private CompletableFuture<Optional<GlobalIndexResult>> unionAsync(
Function<GlobalIndexReader,
CompletableFuture<Optional<GlobalIndexResult>>> visitor) {
long start = durationConsumer == null ? 0L : System.nanoTime();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java
index e9cb833a68..d8183337bd 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java
@@ -25,14 +25,18 @@ import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.globalindex.SortedFileMetaSelector;
import org.apache.paimon.globalindex.SortedIndexFileMeta;
+import org.apache.paimon.globalindex.TopNGlobalIndexResult;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.memory.MemorySegment;
import org.apache.paimon.memory.MemorySlice;
import org.apache.paimon.memory.MemorySliceInput;
+import org.apache.paimon.predicate.SortValue;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.sst.BlockCache;
import org.apache.paimon.sst.BlockHandle;
import org.apache.paimon.sst.BlockIterator;
+import org.apache.paimon.sst.ReverseBlockIterator;
import org.apache.paimon.sst.SstFileReader;
import org.apache.paimon.utils.FileBasedBloomFilter;
import org.apache.paimon.utils.LazyField;
@@ -43,6 +47,7 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
@@ -339,6 +344,20 @@ public class BTreeIndexReader implements Closeable {
return createResult(() -> rangeQuery(from, to, true, true));
}
+ public Optional<GlobalIndexResult> visitTopN(TopN topN) {
+ List<SortValue> orders = topN.orders();
+ if (orders.size() != 1) {
+ return Optional.empty();
+ }
+ Preconditions.checkArgument(topN.limit() >= 0, "TopN limit must not be
negative.");
+ SortValue order = orders.get(0);
+ try {
+ return Optional.of(topN(topN.limit(), order.direction(),
order.nullOrdering()));
+ } catch (IOException e) {
+ throw new RuntimeException("fail to read btree index file.", e);
+ }
+ }
+
private Optional<GlobalIndexResult>
createResult(IOSupplier<RoaringNavigableMap64> supplier) {
try {
return Optional.of(GlobalIndexResult.create(supplier.get()));
@@ -362,6 +381,84 @@ public class BTreeIndexReader implements Closeable {
return rangeQuery(minKey, maxKey, true, true);
}
+ private TopNGlobalIndexResult topN(
+ int limit, SortValue.SortDirection direction,
SortValue.NullOrdering nullOrdering)
+ throws IOException {
+ List<KeyRowIds> result = new ArrayList<>();
+ if (limit == 0) {
+ return TopNGlobalIndexResult.create(result, comparator, direction,
nullOrdering, limit);
+ }
+
+ int remaining = limit;
+ if (nullOrdering == SortValue.NullOrdering.NULLS_FIRST) {
+ remaining = addNullRows(result, remaining);
+ }
+ if (remaining > 0) {
+ remaining =
+ direction == SortValue.SortDirection.ASCENDING
+ ? addAscendingNonNullRows(result, remaining)
+ : addDescendingNonNullRows(result, remaining);
+ }
+ if (remaining > 0 && nullOrdering ==
SortValue.NullOrdering.NULLS_LAST) {
+ addNullRows(result, remaining);
+ }
+ return TopNGlobalIndexResult.create(result, comparator, direction,
nullOrdering, limit);
+ }
+
+ private int addNullRows(List<KeyRowIds> result, int remaining) {
+ int count = (int) Math.min(nullBitmap.get().getLongCardinality(),
remaining);
+ long[] rowIds = new long[count];
+ int position = 0;
+ for (long rowId : nullBitmap.get()) {
+ rowIds[position++] = rowId;
+ if (position == count) {
+ break;
+ }
+ }
+ if (count > 0) {
+ result.add(new KeyRowIds(null, rowIds));
+ }
+ return remaining - count;
+ }
+
+ private int addDescendingNonNullRows(List<KeyRowIds> result, int
remaining) throws IOException {
+ if (maxKey == null) {
+ return remaining;
+ }
+
+ SstFileReader.SstFileReverseIterator fileIterator =
reader.createReverseIterator();
+ ReverseBlockIterator dataIterator;
+ while (remaining > 0 && (dataIterator = fileIterator.readBatch()) !=
null) {
+ while (remaining > 0 && dataIterator.hasNext()) {
+ Map.Entry<MemorySlice, MemorySlice> entry =
dataIterator.next();
+ Object key = keySerializer.deserialize(entry.getKey());
+ long[] rowIds = deserializeRowIds(entry.getValue(), remaining);
+ result.add(new KeyRowIds(key, rowIds));
+ remaining -= rowIds.length;
+ }
+ }
+ return remaining;
+ }
+
+ private int addAscendingNonNullRows(List<KeyRowIds> result, int remaining)
throws IOException {
+ if (minKey == null) {
+ return remaining;
+ }
+
+ SstFileReader.SstFileIterator fileIterator = reader.createIterator();
+ BlockIterator dataIterator;
+ while (remaining > 0 && (dataIterator = fileIterator.readBatch()) !=
null) {
+ while (remaining > 0 && dataIterator.hasNext()) {
+ Map.Entry<MemorySlice, MemorySlice> entry =
dataIterator.next();
+ Object key = keySerializer.deserialize(entry.getKey());
+ long[] rowIds = deserializeRowIds(entry.getValue(), remaining);
+ result.add(new KeyRowIds(key, rowIds));
+ remaining -= rowIds.length;
+ }
+ }
+ return remaining;
+ }
+
/**
* Range query on underlying SST File.
*
@@ -402,11 +499,17 @@ public class BTreeIndexReader implements Closeable {
}
private long[] deserializeRowIds(MemorySlice slice) {
+ return deserializeRowIds(slice, Integer.MAX_VALUE);
+ }
+
+ static long[] deserializeRowIds(MemorySlice slice, int maxRowIds) {
+ Preconditions.checkArgument(maxRowIds >= 0, "Max row id count must not
be negative.");
MemorySliceInput sliceInput = slice.toInput();
int length = sliceInput.readVarLenInt();
Preconditions.checkState(length > 0, "Invalid row id length: 0");
- long[] ids = new long[length];
- for (int i = 0; i < length; i++) {
+ int resultLength = Math.min(length, maxRowIds);
+ long[] ids = new long[resultLength];
+ for (int i = 0; i < resultLength; i++) {
ids[i] = sliceInput.readVarLenLong();
}
return ids;
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 e7e6217e27..d346d868e7 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
@@ -25,11 +25,13 @@ import
org.apache.paimon.globalindex.SortedFileGlobalIndexReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.utils.RoaringNavigableMap64;
import java.io.IOException;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
/**
@@ -147,6 +149,11 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
return bitmap(reader.visitGreaterThan(literal));
}
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitTopN(TopN topN)
{
+ return visitAllFiles(reader -> reader.visitTopN(topN));
+ }
+
@Override
protected BTreeIndexReader openReader(GlobalIndexIOMeta meta) {
try {
diff --git a/paimon-common/src/main/java/org/apache/paimon/sst/BlockReader.java
b/paimon-common/src/main/java/org/apache/paimon/sst/BlockReader.java
index caee5884d5..6c6d0749b1 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sst/BlockReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sst/BlockReader.java
@@ -54,6 +54,10 @@ public abstract class BlockReader {
return new BlockIterator(this);
}
+ public ReverseBlockIterator reverseIterator() {
+ return new ReverseBlockIterator(this);
+ }
+
/** Seek to slice position from record position. */
public abstract int seekTo(int recordPosition);
diff --git
a/paimon-common/src/main/java/org/apache/paimon/sst/ReverseBlockIterator.java
b/paimon-common/src/main/java/org/apache/paimon/sst/ReverseBlockIterator.java
new file mode 100644
index 0000000000..3ed480e595
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/sst/ReverseBlockIterator.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.sst;
+
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.memory.MemorySliceInput;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+/** An {@link Iterator} which reads entries in a block from last to first. */
+public class ReverseBlockIterator implements Iterator<Map.Entry<MemorySlice,
MemorySlice>> {
+
+ private final BlockReader reader;
+ private final MemorySliceInput input;
+ private int recordPosition;
+
+ public ReverseBlockIterator(BlockReader reader) {
+ this.reader = reader;
+ this.input = reader.blockInput();
+ this.recordPosition = reader.recordCount() - 1;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return recordPosition >= 0;
+ }
+
+ @Override
+ public BlockEntry next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+
+ input.setPosition(reader.seekTo(recordPosition--));
+ int keyLength = input.readVarLenInt();
+ MemorySlice key = input.readSlice(keyLength);
+ int valueLength = input.readVarLenInt();
+ MemorySlice value = input.readSlice(valueLength);
+ return new BlockEntry(key, value);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
b/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
index 788e926dc7..21b7a421fa 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
@@ -92,6 +92,10 @@ public class SstFileReader implements Closeable {
return new SstFileIterator(indexBlock.iterator());
}
+ public SstFileReverseIterator createReverseIterator() {
+ return new SstFileReverseIterator(indexBlock.reverseIterator());
+ }
+
private BlockIterator getNextBlock(BlockIterator indexBlockIterator) {
// index block handle, point to the key, value position.
MemorySlice blockHandle = indexBlockIterator.next().getValue();
@@ -100,6 +104,14 @@ public class SstFileReader implements Closeable {
return dataBlock.iterator();
}
+ private ReverseBlockIterator getPreviousBlock(ReverseBlockIterator
indexBlockIterator) {
+ // index block handle, point to the key, value position.
+ MemorySlice blockHandle = indexBlockIterator.next().getValue();
+ BlockReader dataBlock =
+ readBlock(BlockHandle.readBlockHandle(blockHandle.toInput()),
false);
+ return dataBlock.reverseIterator();
+ }
+
/**
* @param blockHandle The block handle.
* @param index Whether read the block as an index.
@@ -214,4 +226,28 @@ public class SstFileReader implements Closeable {
return getNextBlock(indexIterator);
}
}
+
+ /** An iterator which reads an SST file from the largest key to the
smallest key. */
+ public class SstFileReverseIterator {
+
+ private final ReverseBlockIterator indexIterator;
+
+ SstFileReverseIterator(ReverseBlockIterator indexIterator) {
+ this.indexIterator = indexIterator;
+ }
+
+ /**
+ * Read a batch of records from this SST File and move current record
position to the
+ * previous batch.
+ *
+ * @return current batch of records, null if reaching file beginning.
+ */
+ @Nullable
+ public ReverseBlockIterator readBatch() throws IOException {
+ if (!indexIterator.hasNext()) {
+ return null;
+ }
+ return getPreviousBlock(indexIterator);
+ }
+ }
}
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 dcd36c8eaa..42ccc570e2 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
@@ -23,6 +23,7 @@ import org.apache.paimon.predicate.CompoundPredicate;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.predicate.VectorSearch;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
@@ -45,6 +46,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -98,6 +101,46 @@ class GlobalIndexEvaluatorTest {
evaluator.close();
}
+ @Test
+ void testTopNUsesAggregatedReaderAndReusesPredicateCache() {
+ RowType rowType = rowType();
+ AtomicInteger readersCreated = new AtomicInteger();
+ GlobalIndexReader first =
+ new StubGlobalIndexReader(null) {
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitTopN(TopN topN) {
+ return
CompletableFuture.completedFuture(Optional.of(resultOf(1, 2)));
+ }
+ };
+ GlobalIndexReader second =
+ new StubGlobalIndexReader(null) {
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitTopN(TopN topN) {
+ return
CompletableFuture.completedFuture(Optional.of(resultOf(2, 3)));
+ }
+ };
+ GlobalIndexReader union = new
UnionGlobalIndexReader(Arrays.asList(first, second));
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(
+ rowType,
+ fieldId -> {
+ readersCreated.incrementAndGet();
+ return Collections.singletonList(union);
+ });
+ TopN topN = new TopN(new FieldRef(0, "a", DataTypes.INT()),
DESCENDING, NULLS_LAST, 2);
+
+ evaluator.evaluate(new PredicateBuilder(rowType).equal(0, 42));
+ Optional<GlobalIndexResult> firstResult = evaluator.evaluateTopN(topN);
+ Optional<GlobalIndexResult> secondResult =
evaluator.evaluateTopN(topN);
+
+ assertThat(firstResult).isPresent();
+ assertBitmapContainsExactly(firstResult.get().results(), 1L, 2L, 3L);
+ assertThat(secondResult).isPresent();
+ assertBitmapContainsExactly(secondResult.get().results(), 1L, 2L, 3L);
+ assertThat(readersCreated).hasValue(1);
+ evaluator.close();
+ }
+
@Test
void testAndParallelMultipleFields() {
executor = Executors.newFixedThreadPool(2);
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/TopNGlobalIndexResultTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/TopNGlobalIndexResultTest.java
new file mode 100644
index 0000000000..b5e4e00895
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/TopNGlobalIndexResultTest.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.globalindex.btree.BTreeIndexReader.KeyRowIds;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_FIRST;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TopNGlobalIndexResult}. */
+public class TopNGlobalIndexResultTest {
+
+ private static final Comparator<Object> INT_COMPARATOR =
+ (left, right) -> Integer.compare((Integer) left, (Integer) right);
+
+ @Test
+ public void testMergeKeepsGlobalTopN() {
+ TopNGlobalIndexResult first = result(NULLS_LAST, 3, keyRowIds(50, 5),
keyRowIds(10, 1));
+ TopNGlobalIndexResult second = result(NULLS_LAST, 3, keyRowIds(40, 4),
keyRowIds(30, 3));
+
+ GlobalIndexResult merged = first.or(second);
+
+ assertThat(merged).isInstanceOf(TopNGlobalIndexResult.class);
+ assertThat(merged.results()).containsExactlyInAnyOrder(3L, 4L, 5L);
+ }
+
+ @Test
+ public void testMergeKeepsGlobalAscendingTopN() {
+ TopNGlobalIndexResult first =
+ result(ASCENDING, NULLS_LAST, 3, keyRowIds(50, 5),
keyRowIds(10, 1));
+ TopNGlobalIndexResult second =
+ result(ASCENDING, NULLS_LAST, 3, keyRowIds(40, 4),
keyRowIds(30, 3));
+
+ TopNGlobalIndexResult merged = (TopNGlobalIndexResult)
first.or(second);
+
+ assertThat(merged.results()).containsExactlyInAnyOrder(1L, 3L, 4L);
+
assertThat(merged.keyRowIds()).extracting(KeyRowIds::key).containsExactly(10,
30, 40);
+ }
+
+ @Test
+ public void testCannotMergeDifferentDirections() {
+ TopNGlobalIndexResult ascending = result(ASCENDING, NULLS_LAST, 1,
keyRowIds(10, 1));
+ TopNGlobalIndexResult descending = result(DESCENDING, NULLS_LAST, 1,
keyRowIds(10, 1));
+
+ assertThatThrownBy(() -> ascending.or(descending))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Cannot merge sorted global index results with
different sort directions.");
+ }
+
+ @Test
+ public void testMergeBreaksBoundaryTiesByRowId() {
+ TopNGlobalIndexResult first = result(NULLS_LAST, 2, keyRowIds(20, 9),
keyRowIds(10, 3));
+ TopNGlobalIndexResult second = result(NULLS_LAST, 2, keyRowIds(10, 2),
keyRowIds(5, 1));
+
+ assertThat(first.or(second).results()).containsExactlyInAnyOrder(2L,
9L);
+ }
+
+ @Test
+ public void testMergeCombinesSameKeyAndLimitsByRowIdCardinality() {
+ TopNGlobalIndexResult first = result(NULLS_LAST, 4, keyRowIds(20, 9,
3), keyRowIds(10, 1));
+ TopNGlobalIndexResult second = result(NULLS_LAST, 4, keyRowIds(20, 2,
9), keyRowIds(15, 4));
+
+ TopNGlobalIndexResult merged = (TopNGlobalIndexResult)
first.or(second);
+
+ assertThat(merged.results()).containsExactlyInAnyOrder(2L, 3L, 4L, 9L);
+ assertThat(merged.keyRowIds()).hasSize(2);
+ assertThat(merged.keyRowIds().get(0).key()).isEqualTo(20);
+ assertThat(merged.keyRowIds().get(0).rowIds()).containsExactly(2L, 3L,
9L);
+ assertThat(merged.keyRowIds().get(1).key()).isEqualTo(15);
+ assertThat(merged.keyRowIds().get(1).rowIds()).containsExactly(4L);
+ }
+
+ @Test
+ public void testSingleKeyIsLimitedByRowIdCardinality() {
+ TopNGlobalIndexResult result =
+ result(NULLS_LAST, 2, keyRowIds(20, 5, 3, 4), keyRowIds(10,
1));
+
+ assertThat(result.keyRowIds()).hasSize(1);
+ assertThat(result.keyRowIds().get(0).key()).isEqualTo(20);
+ assertThat(result.keyRowIds().get(0).rowIds()).containsExactly(3L, 4L);
+ assertThat(result.results()).containsExactlyInAnyOrder(3L, 4L);
+ }
+
+ @Test
+ public void testNullOrdering() {
+ TopNGlobalIndexResult first = result(NULLS_FIRST, 2, keyRowIds(null,
2), keyRowIds(100, 4));
+ TopNGlobalIndexResult second =
+ result(NULLS_FIRST, 2, keyRowIds(null, 1), keyRowIds(200, 3));
+
+ assertThat(first.or(second).results()).containsExactlyInAnyOrder(1L,
2L);
+
+ first = result(NULLS_LAST, 2, keyRowIds(null, 2), keyRowIds(100, 4));
+ second = result(NULLS_LAST, 2, keyRowIds(null, 1), keyRowIds(200, 3));
+
+ assertThat(first.or(second).results()).containsExactlyInAnyOrder(3L,
4L);
+ }
+
+ @Test
+ public void testOffsetPreservesSortKeys() {
+ TopNGlobalIndexResult first = result(NULLS_LAST, 2, keyRowIds(20, 1),
keyRowIds(10, 2));
+ TopNGlobalIndexResult second = result(NULLS_LAST, 2, keyRowIds(30,
1)).offset(10);
+
+ GlobalIndexResult merged = first.or(second);
+
+ assertThat(merged).isInstanceOf(TopNGlobalIndexResult.class);
+ assertThat(merged.results()).containsExactlyInAnyOrder(1L, 11L);
+ }
+
+ @Test
+ public void testPlainResultUsesConservativeUnion() {
+ TopNGlobalIndexResult sorted = result(NULLS_LAST, 1, keyRowIds(20, 1));
+ RoaringNavigableMap64 unindexedRows = new RoaringNavigableMap64();
+ unindexedRows.add(2);
+
+ GlobalIndexResult merged =
sorted.or(GlobalIndexResult.create(unindexedRows));
+
+ assertThat(merged).isNotInstanceOf(TopNGlobalIndexResult.class);
+ assertThat(merged.results()).containsExactlyInAnyOrder(1L, 2L);
+ }
+
+ private TopNGlobalIndexResult result(
+ org.apache.paimon.predicate.SortValue.NullOrdering nullOrdering,
+ int limit,
+ KeyRowIds... entries) {
+ return result(DESCENDING, nullOrdering, limit, entries);
+ }
+
+ private TopNGlobalIndexResult result(
+ org.apache.paimon.predicate.SortValue.SortDirection direction,
+ org.apache.paimon.predicate.SortValue.NullOrdering nullOrdering,
+ int limit,
+ KeyRowIds... entries) {
+ List<KeyRowIds> candidates = Arrays.asList(entries);
+ return TopNGlobalIndexResult.create(
+ candidates, INT_COMPARATOR, direction, nullOrdering, limit);
+ }
+
+ private KeyRowIds keyRowIds(Integer key, long... rowIds) {
+ return new KeyRowIds(key, rowIds);
+ }
+}
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 3c5e9b07c0..21a9870a79 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
@@ -20,14 +20,24 @@ package org.apache.paimon.globalindex.btree;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.memory.MemorySliceOutput;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.TopN;
import
org.apache.paimon.testutils.junit.parameterized.ParameterizedTestExtension;
+import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
import java.util.Collections;
import java.util.List;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_FIRST;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link BTreeIndexReader} to read a single file. */
@ExtendWith(ParameterizedTestExtension.class)
@@ -43,4 +53,97 @@ public class BTreeIndexReaderTest extends
AbstractIndexReaderTest {
return globalIndexer.createReader(
fileReader, Collections.singletonList(written),
newDirectExecutorService());
}
+
+ @TestTemplate
+ public void testDescendingTopN() throws Exception {
+ int limit = 20;
+ FieldRef ref = new FieldRef(1, "testField", dataType);
+ Object[] valuesByRowId = valuesByRowId();
+
+ try (GlobalIndexReader reader = prepareDataAndCreateReader()) {
+ GlobalIndexResult result =
+ reader.visitTopN(new TopN(ref, DESCENDING, NULLS_LAST,
limit)).join().get();
+ assertThat(result.results().getLongCardinality()).isEqualTo(limit);
+
+ Object boundary = data.get(dataNum - limit).getKey();
+ for (long rowId : result.results()) {
+ assertThat(comparator.compare(valuesByRowId[(int) rowId],
boundary))
+ .isGreaterThanOrEqualTo(0);
+ }
+
+ GlobalIndexResult ascending =
+ reader.visitTopN(new TopN(ref, ASCENDING, NULLS_LAST,
limit)).join().get();
+
assertThat(ascending.results().getLongCardinality()).isEqualTo(limit);
+ boundary = data.get(limit - 1).getKey();
+ for (long rowId : ascending.results()) {
+ assertThat(comparator.compare(valuesByRowId[(int) rowId],
boundary))
+ .isLessThanOrEqualTo(0);
+ }
+
+ assertThat(
+ reader.visitTopN(new TopN(ref, DESCENDING,
NULLS_LAST, 0))
+ .join()
+ .get()
+ .results())
+ .isEmpty();
+ }
+
+ int nullCount = dataNum / 10;
+ for (int i = dataNum - nullCount; i < dataNum; i++) {
+ data.get(i).setLeft(null);
+ }
+ valuesByRowId = valuesByRowId();
+ try (GlobalIndexReader reader = prepareDataAndCreateReader()) {
+ GlobalIndexResult nullsFirst =
+ reader.visitTopN(new TopN(ref, DESCENDING, NULLS_FIRST,
limit)).join().get();
+
assertThat(nullsFirst.results().getLongCardinality()).isEqualTo(limit);
+ for (long rowId : nullsFirst.results()) {
+ assertThat(valuesByRowId[(int) rowId]).isNull();
+ }
+
+ GlobalIndexResult nullsLast =
+ reader.visitTopN(new TopN(ref, DESCENDING, NULLS_LAST,
limit)).join().get();
+
assertThat(nullsLast.results().getLongCardinality()).isEqualTo(limit);
+ Object boundary = data.get(dataNum - nullCount - limit).getKey();
+ for (long rowId : nullsLast.results()) {
+ Object value = valuesByRowId[(int) rowId];
+ assertThat(value).isNotNull();
+ assertThat(comparator.compare(value,
boundary)).isGreaterThanOrEqualTo(0);
+ }
+
+ GlobalIndexResult ascendingNullsFirst =
+ reader.visitTopN(new TopN(ref, ASCENDING, NULLS_FIRST,
limit)).join().get();
+
assertThat(ascendingNullsFirst.results().getLongCardinality()).isEqualTo(limit);
+ for (long rowId : ascendingNullsFirst.results()) {
+ assertThat(valuesByRowId[(int) rowId]).isNull();
+ }
+
+ GlobalIndexResult ascendingNullsLast =
+ reader.visitTopN(new TopN(ref, ASCENDING, NULLS_LAST,
limit)).join().get();
+
assertThat(ascendingNullsLast.results().getLongCardinality()).isEqualTo(limit);
+ boundary = data.get(limit - 1).getKey();
+ for (long rowId : ascendingNullsLast.results()) {
+ Object value = valuesByRowId[(int) rowId];
+ assertThat(value).isNotNull();
+ assertThat(comparator.compare(value,
boundary)).isLessThanOrEqualTo(0);
+ }
+ }
+ }
+
+ @TestTemplate
+ public void testTopNOnlyDeserializesRemainingRowIds() {
+ MemorySliceOutput output = new MemorySliceOutput(16);
+ output.writeVarLenInt(3);
+ output.writeVarLenLong(10);
+ output.writeVarLenLong(20);
+
+ assertThat(BTreeIndexReader.deserializeRowIds(output.toSlice(), 2))
+ .containsExactly(10L, 20L);
+ }
+
+ private Object[] valuesByRowId() {
+ Object[] values = new Object[dataNum];
+ data.forEach(pair -> values[pair.getValue().intValue()] =
pair.getKey());
+ return values;
+ }
}
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 760c5daa9f..d8321d7684 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
@@ -24,9 +24,11 @@ import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.btree.BTreeIndexReader.KeyRowIds;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.TopN;
import
org.apache.paimon.testutils.junit.parameterized.ParameterizedTestExtension;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.Pair;
@@ -52,6 +54,9 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
import static org.assertj.core.api.Assertions.assertThat;
@@ -93,6 +98,37 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
return -1;
}
+ @TestTemplate
+ public void testGlobalTopNCandidatesAcrossFiles() throws Exception {
+ int limit = 5;
+ List<GlobalIndexIOMeta> written = writeData();
+ FieldRef ref = new FieldRef(1, "testField", dataType);
+
+ try (GlobalIndexReader reader =
+ globalIndexer.createReader(fileReader, written,
newDirectExecutorService())) {
+ GlobalIndexResult result =
+ reader.visitTopN(new TopN(ref, DESCENDING, NULLS_LAST,
limit)).join().get();
+ assertThat(result.results().getLongCardinality()).isEqualTo(limit);
+
+ Object boundary = data.get(dataNum - limit).getKey();
+ Object[] valuesByRowId = new Object[dataNum];
+ data.forEach(pair -> valuesByRowId[pair.getValue().intValue()] =
pair.getKey());
+ for (long rowId : result.results()) {
+ assertThat(comparator.compare(valuesByRowId[(int) rowId],
boundary))
+ .isGreaterThanOrEqualTo(0);
+ }
+
+ GlobalIndexResult ascending =
+ reader.visitTopN(new TopN(ref, ASCENDING, NULLS_LAST,
limit)).join().get();
+
assertThat(ascending.results().getLongCardinality()).isEqualTo(limit);
+ boundary = data.get(limit - 1).getKey();
+ for (long rowId : ascending.results()) {
+ assertThat(comparator.compare(valuesByRowId[(int) rowId],
boundary))
+ .isLessThanOrEqualTo(0);
+ }
+ }
+ }
+
@TestTemplate
public void testFallbackScanDisabledByBudget() throws Exception {
options.set(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE,
MemorySize.ofBytes(1));
@@ -185,7 +221,7 @@ public class LazyFilteredBTreeIndexReaderTest extends
AbstractIndexReaderTest {
// Collect all entries from iterator
while (iter.hasNext()) {
- BTreeIndexReader.KeyRowIds entry = iter.next();
+ KeyRowIds entry = iter.next();
Object key = entry.key();
long[] rowIds = entry.rowIds();
diff --git
a/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
b/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
index 92e6ae8d2e..f40b464399 100644
--- a/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
@@ -47,7 +47,8 @@ public class BlockIteratorTest {
public void innerTest(boolean aligned) throws IOException {
MemorySlice data = writeBlock(aligned);
- BlockIterator iterator = BlockReader.create(data,
COMPARATOR).iterator();
+ BlockReader reader = BlockReader.create(data, COMPARATOR);
+ BlockIterator iterator = reader.iterator();
// 1. test for normal cases:
final int step = 3;
@@ -92,6 +93,18 @@ public class BlockIteratorTest {
Assertions.assertTrue(iterator.hasNext());
entry = iterator.next();
Assertions.assertEquals(4, entry.getKey().readInt(0));
+
+ ReverseBlockIterator reverseIterator = reader.reverseIterator();
+ int expected = ROW_NUM - 1;
+ while (reverseIterator.hasNext()) {
+ Map.Entry<MemorySlice, MemorySlice> reverseEntry =
reverseIterator.next();
+ Assertions.assertEquals(expected * 2,
reverseEntry.getKey().readInt(0));
+ Assertions.assertArrayEquals(
+ constructValue(valueOut, aligned, expected),
+ reverseEntry.getValue().copyBytes());
+ expected--;
+ }
+ Assertions.assertEquals(-1, expected);
}
private MemorySlice writeBlock(boolean aligned) throws IOException {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/BTreeTopNIndexFileSelector.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/BTreeTopNIndexFileSelector.java
new file mode 100644
index 0000000000..69b52276c4
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/BTreeTopNIndexFileSelector.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.predicate.SortValue;
+import org.apache.paimon.predicate.TopN;
+import org.apache.paimon.types.DataField;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * Selects BTree index files which may contain a single-column TopN result.
+ *
+ * <p>Every BTree file must have sorted metadata, matching the predicate
reader contract. Retaining
+ * the first {@code N} files ordered by their best value is safe because every
BTree file
+ * contributes at least one row at that value. Fewer files can be retained
when one file alone
+ * contains at least {@code N} rows and its worst value is not worse than the
best value of every
+ * remaining file.
+ */
+class BTreeTopNIndexFileSelector {
+
+ private final KeySerializer keySerializer;
+ private final Comparator<Object> keyComparator;
+ private final boolean ascending;
+ private final boolean nullsFirst;
+
+ private BTreeTopNIndexFileSelector(DataField field, TopN topN) {
+ this.keySerializer = KeySerializer.create(field.type());
+ this.keyComparator = keySerializer.createComparator();
+ this.ascending = topN.orders().get(0).direction() ==
SortValue.SortDirection.ASCENDING;
+ this.nullsFirst = topN.orders().get(0).nullOrdering() ==
SortValue.NullOrdering.NULLS_FIRST;
+ }
+
+ static List<IndexFileMeta> select(List<IndexFileMeta> files, DataField
field, TopN topN) {
+ int limit = topN.limit();
+ if (limit == 0) {
+ return new ArrayList<>();
+ }
+
+ BTreeTopNIndexFileSelector selector = new
BTreeTopNIndexFileSelector(field, topN);
+ List<IndexFileMeta> selected = new ArrayList<>();
+ List<RankedIndexFile> rankedFiles = new ArrayList<>();
+ for (IndexFileMeta file : files) {
+ rankedFiles.add(selector.rank(file));
+ }
+
+ rankedFiles.sort(selector::compare);
+ for (int i = 0; i < Math.min(limit, rankedFiles.size()); i++) {
+ RankedIndexFile current = rankedFiles.get(i);
+ selected.add(current.file);
+ if (i + 1 < rankedFiles.size()
+ && current.file.rowCount() >= limit
+ // Equal boundary keys are safe because this TopN has no
secondary ordering or
+ // WITH TIES semantics, and the current file alone
supplies enough rows.
+ && selector.compareWorstToBest(current, rankedFiles.get(i
+ 1)) <= 0) {
+ break;
+ }
+ }
+ return selected;
+ }
+
+ private RankedIndexFile rank(IndexFileMeta file) {
+ GlobalIndexMeta globalIndex =
+ checkNotNull(
+ file.globalIndexMeta(),
+ "BTree index file '%s' is missing global index
metadata.",
+ file.fileName());
+ byte[] indexMeta =
+ checkNotNull(
+ globalIndex.indexMeta(),
+ "BTree index file '%s' is missing sorted metadata.",
+ file.fileName());
+ SortedIndexFileMeta sortedMeta =
SortedIndexFileMeta.deserialize(indexMeta);
+ byte[] firstKey = sortedMeta.firstKey();
+ byte[] lastKey = sortedMeta.lastKey();
+ boolean hasNonNulls = lastKey != null;
+
+ byte[] nonNullBestKey = ascending ? firstKey : lastKey;
+ byte[] nonNullWorstKey = ascending ? lastKey : firstKey;
+ boolean bestIsNull = nullsFirst ? sortedMeta.hasNulls() : !hasNonNulls;
+ Object bestKey =
+ bestIsNull ? null :
keySerializer.deserialize(MemorySlice.wrap(nonNullBestKey));
+ boolean worstIsNull = nullsFirst ? !hasNonNulls :
sortedMeta.hasNulls();
+ Object worstKey =
+ worstIsNull ? null :
keySerializer.deserialize(MemorySlice.wrap(nonNullWorstKey));
+ return new RankedIndexFile(file, bestIsNull, bestKey, worstIsNull,
worstKey);
+ }
+
+ private int compare(RankedIndexFile left, RankedIndexFile right) {
+ int result = compareValues(left.bestIsNull, left.bestKey,
right.bestIsNull, right.bestKey);
+ if (result != 0) {
+ return result;
+ }
+ return left.file.fileName().compareTo(right.file.fileName());
+ }
+
+ private int compareWorstToBest(RankedIndexFile current, RankedIndexFile
next) {
+ return compareValues(current.worstIsNull, current.worstKey,
next.bestIsNull, next.bestKey);
+ }
+
+ private int compareValues(
+ boolean leftIsNull,
+ @Nullable Object left,
+ boolean rightIsNull,
+ @Nullable Object right) {
+ if (leftIsNull != rightIsNull) {
+ if (leftIsNull) {
+ return nullsFirst ? -1 : 1;
+ }
+ return nullsFirst ? 1 : -1;
+ }
+ if (leftIsNull) {
+ return 0;
+ }
+ return ascending ? keyComparator.compare(left, right) :
keyComparator.compare(right, left);
+ }
+
+ private static class RankedIndexFile {
+
+ private final IndexFileMeta file;
+ private final boolean bestIsNull;
+ @Nullable private final Object bestKey;
+ private final boolean worstIsNull;
+ @Nullable private final Object worstKey;
+
+ private RankedIndexFile(
+ IndexFileMeta file,
+ boolean bestIsNull,
+ @Nullable Object bestKey,
+ boolean worstIsNull,
+ @Nullable Object worstKey) {
+ this.file = file;
+ this.bestIsNull = bestIsNull;
+ this.bestKey = bestKey;
+ this.worstIsNull = worstIsNull;
+ this.worstKey = worstKey;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
index 6dbbca8e77..059d661608 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
@@ -68,6 +68,8 @@ public class DataEvolutionBatchScan implements DataTableScan {
private final AppendBatchTableScan batchScan;
private Predicate filter;
+ private TopN topN;
+ private Integer pushDownLimit;
private RowRangeIndex pushedRowRangeIndex;
private GlobalIndexResult globalIndexResult;
@@ -148,7 +150,7 @@ public class DataEvolutionBatchScan implements
DataTableScan {
@Override
public InnerTableScan withTopN(TopN topN) {
- batchScan.withTopN(topN);
+ this.topN = topN;
return this;
}
@@ -166,6 +168,7 @@ public class DataEvolutionBatchScan implements
DataTableScan {
@Override
public InnerTableScan withLimit(int limit) {
+ this.pushDownLimit = limit;
batchScan.withLimit(limit);
return this;
}
@@ -255,9 +258,16 @@ public class DataEvolutionBatchScan implements
DataTableScan {
public Plan plan() {
RowRangeIndex rowRangeIndex = this.pushedRowRangeIndex;
ScoreGetter scoreGetter = null;
+ boolean globalIndexTopNCandidatesFound = false;
if (rowRangeIndex == null) {
- Optional<GlobalIndexResult> indexResult = evalGlobalIndex();
+ Optional<GlobalIndexResult> indexResult;
+ if (canPushDownGlobalIndexTopN()) {
+ indexResult = evalGlobalIndexTopN();
+ globalIndexTopNCandidatesFound = indexResult.isPresent();
+ } else {
+ indexResult = evalGlobalIndex();
+ }
if (indexResult.isPresent()) {
GlobalIndexResult result = indexResult.get();
rowRangeIndex =
RowRangeIndex.create(result.results().toRangeList());
@@ -269,6 +279,10 @@ public class DataEvolutionBatchScan implements
DataTableScan {
}
}
+ if (!globalIndexTopNCandidatesFound && topN != null) {
+ batchScan.withTopN(topN);
+ }
+
if (rowRangeIndex == null) {
return batchScan.plan();
}
@@ -328,6 +342,73 @@ public class DataEvolutionBatchScan implements
DataTableScan {
}
}
+ private Optional<GlobalIndexResult> evalGlobalIndexTopN() {
+ CoreOptions options = table.coreOptions();
+ PartitionPredicate partitionFilter =
+ batchScan.snapshotReader().manifestsReader().partitionFilter();
+ long totalStart = System.nanoTime();
+ Optional<DataEvolutionGlobalIndexScanner> optionalScanner =
+ DataEvolutionGlobalIndexScanner.createForTopN(table,
partitionFilter, topN);
+ long metadataDuration = System.nanoTime() - totalStart;
+ if (!optionalScanner.isPresent()) {
+ return Optional.empty();
+ }
+
+ try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
+ long lookupStart = System.nanoTime();
+ Optional<GlobalIndexResult> result = scanner.scan(topN);
+ long lookupDuration = System.nanoTime() - lookupStart;
+ if (!result.isPresent()) {
+ return Optional.empty();
+ }
+
+ long coverageStart = System.nanoTime();
+ GlobalIndexResult finalResult =
result.get().or(scanner.unindexedRows(topN));
+ long coverageDuration = System.nanoTime() - coverageStart;
+ long totalDuration = System.nanoTime() - totalStart;
+ LOG.info(
+ "Scan table '{}' with BTree global index TopN.
searchMode='{}', topN='{}', total={} ms, metadata={} ms, lookup={} ms,
coverage={} ms.",
+ table.name(),
+ options.scalarIndexSearchMode(),
+ topN,
+ totalDuration / 1_000_000,
+ metadataDuration / 1_000_000,
+ lookupDuration / 1_000_000,
+ coverageDuration / 1_000_000);
+ return Optional.of(finalResult);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private boolean canPushDownGlobalIndexTopN() {
+ if (topN == null
+ || pushDownLimit != null
+ || globalIndexResult != null
+ ||
!table.rowType().containsField(topN.orders().get(0).field().name())) {
+ return false;
+ }
+ CoreOptions options = table.coreOptions();
+ return supportsGlobalIndexTopN(options)
+ && options.globalIndexEnabled()
+ && !options.deletionVectorsEnabled()
+ && !options.queryAuthEnabled()
+ && !batchScan.snapshotReader().hasNonPartitionFilter();
+ }
+
+ private boolean supportsGlobalIndexTopN(CoreOptions options) {
+ switch (options.startupMode()) {
+ case LATEST_FULL:
+ case LATEST:
+ case FROM_TIMESTAMP:
+ case FROM_SNAPSHOT:
+ case FROM_SNAPSHOT_FULL:
+ return true;
+ default:
+ return false;
+ }
+ }
+
@VisibleForTesting
public static Plan wrapToIndexSplits(
List<Split> splits, RowRangeIndex rowRangeIndex, ScoreGetter
scoreGetter) {
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 7dbdb06853..498130a11d 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
@@ -21,6 +21,7 @@ package org.apache.paimon.globalindex;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
@@ -29,6 +30,7 @@ import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
@@ -66,6 +68,8 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
/** Scanner for shard-based global indexes on data-evolution tables. */
public class DataEvolutionGlobalIndexScanner implements Closeable {
+ private static final int MAX_TOP_N_LIMIT = 100;
+
private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionGlobalIndexScanner.class);
@@ -86,6 +90,28 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
FileIO fileIO,
IndexPathFactory indexPathFactory,
Collection<IndexFileMeta> indexFiles) {
+ this(
+ table,
+ snapshot,
+ partitionFilter,
+ options,
+ rowType,
+ fileIO,
+ indexPathFactory,
+ indexFiles,
+ indexFiles);
+ }
+
+ private DataEvolutionGlobalIndexScanner(
+ FileStoreTable table,
+ @Nullable Snapshot snapshot,
+ @Nullable PartitionPredicate partitionFilter,
+ Options options,
+ RowType rowType,
+ FileIO fileIO,
+ IndexPathFactory indexPathFactory,
+ Collection<IndexFileMeta> coverageIndexFiles,
+ Collection<IndexFileMeta> indexFiles) {
this.table = table;
this.options = options;
this.rowType = rowType;
@@ -97,7 +123,7 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
table,
snapshot,
partitionFilter,
- indexFiles,
+ coverageIndexFiles,
table.coreOptions().scalarIndexSearchMode());
GlobalIndexFileReader indexFileReader = meta ->
fileIO.newInputStream(meta.filePath());
Map<Integer, IndexMetaFileGroup> indexMetas = new HashMap<>();
@@ -128,7 +154,6 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
}
group.addFile(indexType, range, indexFile);
}
-
IntFunction<Collection<GlobalIndexReader>> readersFunction =
fId -> {
List<IndexMetaFileGroup> groups = new ArrayList<>();
@@ -244,6 +269,65 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
indexFiles));
}
+ /**
+ * Creates a scanner for a single-column TopN backed by a primary BTree
index.
+ *
+ * <p>Indexes carrying the ordered field only as an extra field are not
ordered by that field
+ * and cannot serve this scan.
+ */
+ public static Optional<DataEvolutionGlobalIndexScanner> createForTopN(
+ FileStoreTable table, @Nullable PartitionPredicate
partitionFilter, TopN topN) {
+ if (!isSupportedTopN(topN)) {
+ return Optional.empty();
+ }
+
+ DataField indexField =
table.rowType().getField(topN.orders().get(0).field().name());
+ int fieldId = indexField.id();
+ @Nullable Snapshot snapshot = tryTravelOrLatest(table);
+ List<IndexFileMeta> indexFiles =
+ table.store().newIndexFileHandler()
+ .scan(snapshot, topNIndexFileFilter(partitionFilter,
fieldId)).stream()
+ .map(IndexManifestEntry::indexFile)
+ .collect(Collectors.toList());
+ if (indexFiles.isEmpty()) {
+ return Optional.empty();
+ }
+ List<IndexFileMeta> selectedIndexFiles =
+ BTreeTopNIndexFileSelector.select(indexFiles, indexField,
topN);
+ return Optional.of(
+ new DataEvolutionGlobalIndexScanner(
+ table,
+ snapshot,
+ partitionFilter,
+ table.coreOptions().toConfiguration(),
+ table.rowType(),
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ indexFiles,
+ selectedIndexFiles));
+ }
+
+ private static boolean isSupportedTopN(TopN topN) {
+ return topN != null
+ && topN.limit() >= 0
+ && topN.limit() <= MAX_TOP_N_LIMIT
+ && topN.orders().size() == 1;
+ }
+
+ private static Filter<IndexManifestEntry> topNIndexFileFilter(
+ @Nullable PartitionPredicate partitionFilter, int fieldId) {
+ return entry -> {
+ if (partitionFilter != null &&
!partitionFilter.test(entry.partition())) {
+ return false;
+ }
+ IndexFileMeta indexFile = entry.indexFile();
+ GlobalIndexMeta globalIndex = indexFile.globalIndexMeta();
+ return globalIndex != null
+ && globalIndex.indexFieldId() == fieldId
+ &&
BTreeGlobalIndexerFactory.IDENTIFIER.equals(indexFile.indexType());
+ };
+ }
+
private static Filter<IndexManifestEntry> indexFileFilter(
FileStoreTable table,
@Nullable PartitionPredicate partitionFilter,
@@ -288,6 +372,20 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
return globalIndexEvaluator.evaluate(predicate);
}
+ public Optional<GlobalIndexResult> scan(TopN topN) {
+ if (!isSupportedTopN(topN)) {
+ return Optional.empty();
+ }
+ if (topN.limit() == 0) {
+ return Optional.of(GlobalIndexResult.createEmpty());
+ }
+ String fieldName = topN.orders().get(0).field().name();
+ if (!rowType.containsField(fieldName)) {
+ return Optional.empty();
+ }
+ return globalIndexEvaluator.evaluateTopN(topN);
+ }
+
public GlobalIndexResult unindexedRows(Predicate predicate) {
RoaringNavigableMap64 rows = new RoaringNavigableMap64();
for (Range range : coverage.unindexedRanges(rowType, predicate)) {
@@ -296,6 +394,15 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
return GlobalIndexResult.create(rows);
}
+ public GlobalIndexResult unindexedRows(TopN topN) {
+ String fieldName = topN.orders().get(0).field().name();
+ RoaringNavigableMap64 rows = new RoaringNavigableMap64();
+ for (Range range :
coverage.unindexedRanges(rowType.getField(fieldName).id())) {
+ rows.addRange(range);
+ }
+ return GlobalIndexResult.create(rows);
+ }
+
private Collection<GlobalIndexReader> createReaders(
GlobalIndexFileReader indexFileReadWrite, IndexMetaFileGroup
group, RowType rowType) {
DataField indexField = group.indexField(rowType);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/BTreeTopNIndexFileSelectorTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/BTreeTopNIndexFileSelectorTest.java
new file mode 100644
index 0000000000..044f286345
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/BTreeTopNIndexFileSelectorTest.java
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.SortValue;
+import org.apache.paimon.predicate.TopN;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_FIRST;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link BTreeTopNIndexFileSelector}. */
+public class BTreeTopNIndexFileSelectorTest {
+
+ private static final DataField FIELD = new DataField(1, "score",
DataTypes.INT());
+ private static final KeySerializer KEY_SERIALIZER =
KeySerializer.create(FIELD.type());
+
+ @Test
+ public void testSelectDescendingNullsLast() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("all-null", null, null, true),
+ file("max-20", 10, 20, false),
+ file("max-40", 30, 40, false),
+ file("max-30", 20, 30, true));
+
+ assertThat(fileNames(select(files, NULLS_LAST,
2))).containsExactly("max-40", "max-30");
+ }
+
+ @Test
+ public void testSelectDescendingNullsFirst() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("nonnull", 100, 1000, false),
+ file("null-b", null, null, true),
+ file("null-a", 10, 20, true));
+
+ assertThat(fileNames(select(files, NULLS_FIRST,
2))).containsExactly("null-a", "null-b");
+ }
+
+ @Test
+ public void testSelectAscendingNullsLast() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("all-null", null, null, true),
+ file("min-10", 10, 20, false),
+ file("min-30", 30, 40, false),
+ file("min-20", 20, 30, true));
+
+ assertThat(fileNames(select(files, ASCENDING, NULLS_LAST, 2)))
+ .containsExactly("min-10", "min-20");
+ }
+
+ @Test
+ public void testSelectAscendingNullsFirst() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("nonnull", 1, 100, false),
+ file("null-b", null, null, true),
+ file("null-a", 10, 20, true));
+
+ assertThat(fileNames(select(files, ASCENDING, NULLS_FIRST, 2)))
+ .containsExactly("null-a", "null-b");
+ }
+
+ @Test
+ public void testAscendingSingleFileCoversTopNAtEqualBoundary() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("bottom", 0, 10, false, 100),
+ file("next", 10, 20, false, 100),
+ file("upper", 20, 30, false, 100));
+
+ assertThat(fileNames(select(files, ASCENDING, NULLS_LAST,
100))).containsExactly("bottom");
+ }
+
+ @Test
+ public void testMissingMetadataFailsFast() {
+ assertThatThrownBy(
+ () ->
select(Arrays.asList(fileWithoutMetadata("missing")), NULLS_LAST, 1))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("BTree index file 'missing' is missing sorted
metadata.");
+ }
+
+ @Test
+ public void testCorruptMetadataFailsFast() {
+ assertThatThrownBy(
+ () ->
+ select(
+ Arrays.asList(
+ fileWithMetadata(
+ "corrupt", new byte[]
{-1, -1, -1, -1})),
+ NULLS_LAST,
+ 1))
+ .isInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ public void testSingleFileCoversTopNAtEqualBoundary() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("top", 90, 100, false, 100),
+ file("next", 80, 90, false, 100),
+ file("lower", 70, 80, false, 100));
+
+ assertThat(fileNames(select(files, NULLS_LAST,
100))).containsExactly("top");
+ }
+
+ @Test
+ public void testDoesNotStopWhenSingleFileHasTooFewRows() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("top", 90, 100, false, 99),
+ file("next", 80, 89, false, 100),
+ file("lower", 70, 79, false, 100));
+
+ assertThat(fileNames(select(files, NULLS_LAST,
100))).containsExactly("top", "next");
+ }
+
+ @Test
+ public void testDoesNotStopAtOverlappingRange() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("top", 0, 100, false, 100),
+ file("next", 90, 99, false, 99),
+ file("lower", 80, 89, false, 99));
+
+ assertThat(fileNames(select(files, NULLS_LAST, 100)))
+ .containsExactly("top", "next", "lower");
+ }
+
+ @Test
+ public void testNullsLastPreventsWholeFileCoverage() {
+ List<IndexFileMeta> files =
+ Arrays.asList(
+ file("top", 90, 100, true, 100),
+ file("next", 80, 89, false, 100),
+ file("lower", 70, 79, false, 100));
+
+ assertThat(fileNames(select(files, NULLS_LAST,
100))).containsExactly("top", "next");
+ }
+
+ @Test
+ public void testLimitBoundaries() {
+ List<IndexFileMeta> files =
+ Arrays.asList(file("max-10", 0, 10, false), file("max-20", 10,
20, false));
+
+ assertThat(select(files, NULLS_LAST, 0)).isEmpty();
+ assertThat(fileNames(select(files, NULLS_LAST, files.size())))
+ .containsExactly("max-20", "max-10");
+ }
+
+ private List<IndexFileMeta> select(
+ List<IndexFileMeta> files, SortValue.NullOrdering nullOrdering,
int limit) {
+ return select(files, DESCENDING, nullOrdering, limit);
+ }
+
+ private List<IndexFileMeta> select(
+ List<IndexFileMeta> files,
+ SortValue.SortDirection direction,
+ SortValue.NullOrdering nullOrdering,
+ int limit) {
+ FieldRef fieldRef = new FieldRef(FIELD.id(), FIELD.name(),
FIELD.type());
+ TopN topN = new TopN(fieldRef, direction, nullOrdering, limit);
+ return BTreeTopNIndexFileSelector.select(files, FIELD, topN);
+ }
+
+ private IndexFileMeta file(
+ String fileName, Integer firstKey, Integer lastKey, boolean
hasNulls) {
+ return file(fileName, firstKey, lastKey, hasNulls, 1);
+ }
+
+ private IndexFileMeta file(
+ String fileName, Integer firstKey, Integer lastKey, boolean
hasNulls, long rowCount) {
+ SortedIndexFileMeta sortedMeta =
+ new SortedIndexFileMeta(serialize(firstKey),
serialize(lastKey), hasNulls);
+ return new IndexFileMeta(
+ "btree",
+ fileName,
+ 1,
+ rowCount,
+ new GlobalIndexMeta(0, 0, FIELD.id(), null,
sortedMeta.serialize()),
+ null);
+ }
+
+ private IndexFileMeta fileWithoutMetadata(String fileName) {
+ return fileWithMetadata(fileName, null);
+ }
+
+ private IndexFileMeta fileWithMetadata(String fileName, byte[] metadata) {
+ return new IndexFileMeta(
+ "btree",
+ fileName,
+ 1,
+ 1,
+ new GlobalIndexMeta(0, 0, FIELD.id(), null, metadata),
+ null);
+ }
+
+ private byte[] serialize(Integer value) {
+ return value == null ? null : KEY_SERIALIZER.serialize(value);
+ }
+
+ private List<String> fileNames(List<IndexFileMeta> files) {
+ return
files.stream().map(IndexFileMeta::fileName).collect(Collectors.toList());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
index 676a1c305e..375cac3989 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
@@ -33,8 +33,10 @@ import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
@@ -63,6 +65,9 @@ import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -146,6 +151,251 @@ public class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
assertThat(readF1).containsExactly("a200", "a300", "a400", "a56789");
}
+ @Test
+ public void testBTreeGlobalIndexTopNCandidatesAcrossRanges() throws
Exception {
+ write(100L);
+ createIndex("f1");
+ appendRows(100, 200);
+ createIndexIncremental("f1");
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
+ TopN topN =
+ new TopN(
+ new FieldRef(1, "f1", table.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 5);
+
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.createForTopN(
+ table, PartitionPredicate.ALWAYS_TRUE, topN)
+ .orElseThrow(AssertionError::new)) {
+
assertThat(scanner.scan(topN).orElseThrow(AssertionError::new).results().toRangeList())
+ .containsExactly(new Range(95, 99));
+ }
+
+ ReadBuilder readBuilder = table.newReadBuilder().withTopN(topN);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+ assertThat(plan.splits()).allMatch(IndexedSplit.class::isInstance);
+ assertThat(readF1(readBuilder, plan))
+ .containsExactlyInAnyOrder("a95", "a96", "a97", "a98", "a99");
+
+ TopN ascendingTopN =
+ new TopN(
+ new FieldRef(1, "f1", table.rowType().getTypeAt(1)),
+ ASCENDING,
+ NULLS_LAST,
+ 5);
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.createForTopN(
+ table, PartitionPredicate.ALWAYS_TRUE,
ascendingTopN)
+ .orElseThrow(AssertionError::new)) {
+ assertThat(
+ scanner.scan(ascendingTopN)
+ .orElseThrow(AssertionError::new)
+ .results()
+ .toRangeList())
+ .containsExactly(new Range(0, 1), new Range(10, 10), new
Range(100, 101));
+ }
+
+ ReadBuilder ascendingReadBuilder =
table.newReadBuilder().withTopN(ascendingTopN);
+ TableScan.Plan ascendingPlan = ascendingReadBuilder.newScan().plan();
+
assertThat(ascendingPlan.splits()).allMatch(IndexedSplit.class::isInstance);
+ assertThat(readF1(ascendingReadBuilder, ascendingPlan))
+ .containsExactlyInAnyOrder("a0", "a1", "a10", "a100", "a101");
+ }
+
+ @Test
+ public void testBTreeGlobalIndexTopNCandidatesSkipSplitTopN() throws
Exception {
+ write(100L);
+ createIndex("f0");
+ appendRows(100, 200);
+ createIndexIncremental("f0");
+
+ FileStoreTable table =
+ ((FileStoreTable) catalog.getTable(identifier()))
+ .copy(
+ Collections.singletonMap(
+
CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), "1 b"));
+ TopN topN =
+ new TopN(
+ new FieldRef(0, "f0", table.rowType().getTypeAt(0)),
+ DESCENDING,
+ NULLS_LAST,
+ 1);
+
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.createForTopN(
+ table, PartitionPredicate.ALWAYS_TRUE, topN)
+ .orElseThrow(AssertionError::new)) {
+
assertThat(scanner.scan(topN).orElseThrow(AssertionError::new).results().toRangeList())
+ .containsExactly(new Range(199, 199));
+ // Index-file TopN pruning must not make the excluded indexed
range look unindexed.
+ assertThat(scanner.unindexedRows(topN).results()).isEmpty();
+ }
+
+ TableScan.Plan plan =
table.newReadBuilder().withTopN(topN).newScan().plan();
+ assertThat(plan.splits()).allMatch(IndexedSplit.class::isInstance);
+ assertThat(
+ plan.splits().stream()
+ .map(IndexedSplit.class::cast)
+ .flatMap(split -> split.rowRanges().stream())
+ .collect(Collectors.toList()))
+ .containsExactly(new Range(199, 199));
+ }
+
+ @Test
+ public void testBTreeGlobalIndexTopNPartialCoverage() throws Exception {
+ write(100L);
+ createIndex("f1");
+ appendRows(100, 110);
+
+ FileStoreTable table =
+ tableWithSearchMode((FileStoreTable)
catalog.getTable(identifier()), "full");
+ TopN topN =
+ new TopN(
+ new FieldRef(1, "f1", table.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 5);
+
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.createForTopN(
+ table, PartitionPredicate.ALWAYS_TRUE, topN)
+ .orElseThrow(AssertionError::new)) {
+
assertThat(scanner.scan(topN).orElseThrow(AssertionError::new).results().toRangeList())
+ .containsExactly(new Range(95, 99));
+ assertThat(scanner.unindexedRows(topN).results().toRangeList())
+ .containsExactly(new Range(100, 109));
+ }
+
+ FileStoreTable fastTable = tableWithSearchMode(table, "fast");
+ try (DataEvolutionGlobalIndexScanner scanner =
+ DataEvolutionGlobalIndexScanner.createForTopN(
+ fastTable, PartitionPredicate.ALWAYS_TRUE,
topN)
+ .orElseThrow(AssertionError::new)) {
+ assertThat(scanner.unindexedRows(topN).results()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testBTreeGlobalIndexTopNFallsBackForUnsafeReads() throws
Exception {
+ write(100L);
+ createIndex("f1");
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
+ TopN topN =
+ new TopN(
+ new FieldRef(1, "f1", table.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 5);
+ Predicate filter = new PredicateBuilder(table.rowType()).lessThan(0,
10);
+ ReadBuilder filtered =
table.newReadBuilder().withFilter(filter).withTopN(topN);
+ TableScan.Plan filteredPlan = filtered.newScan().plan();
+
assertThat(filteredPlan.splits()).allMatch(DataSplit.class::isInstance);
+ assertThat(readF1(filtered, filteredPlan))
+ .containsExactlyInAnyOrder(
+ "a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7", "a8",
"a9");
+
+ FileStoreTable modifiableDeletionVectorTable =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.DELETION_VECTORS_MODIFIABLE.key(),
"true"));
+ FileStoreTable deletionVectorTable =
+ modifiableDeletionVectorTable.copy(
+ Collections.singletonMap(
+ CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true"));
+ ReadBuilder deletionVectorRead =
deletionVectorTable.newReadBuilder().withTopN(topN);
+ assertThat(deletionVectorRead.newScan().plan().splits())
+ .allMatch(DataSplit.class::isInstance);
+ }
+
+ @Test
+ public void testBTreeGlobalIndexTopNFallsBackForLargeLimit() throws
Exception {
+ write(200L);
+ createIndex("f1");
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
+ TopN maxSupportedTopN =
+ new TopN(
+ new FieldRef(1, "f1", table.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 100);
+
assertThat(table.newReadBuilder().withTopN(maxSupportedTopN).newScan().plan().splits())
+ .isNotEmpty()
+ .allMatch(IndexedSplit.class::isInstance);
+
+ TopN topN =
+ new TopN(
+ new FieldRef(1, "f1", table.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 101);
+
+ assertThat(
+ DataEvolutionGlobalIndexScanner.createForTopN(
+ table, PartitionPredicate.ALWAYS_TRUE, topN))
+ .isEmpty();
+
+ TableScan.Plan plan =
table.newReadBuilder().withTopN(topN).newScan().plan();
+
assertThat(plan.splits()).isNotEmpty().allMatch(DataSplit.class::isInstance);
+ }
+
+ @Test
+ public void testBTreeGlobalIndexTopNFallsBackForUnsupportedStartupModes()
throws Exception {
+ write(100L);
+ createIndex("f1");
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
+ long startSnapshot = table.snapshotManager().latestSnapshotId();
+ appendRows(100, 110);
+ table = (FileStoreTable) catalog.getTable(identifier());
+ long endSnapshot = table.snapshotManager().latestSnapshotId();
+
+ FileStoreTable incrementalTable =
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.INCREMENTAL_BETWEEN.key(),
+ startSnapshot + "," + endSnapshot));
+ TopN topN =
+ new TopN(
+ new FieldRef(1, "f1",
incrementalTable.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 5);
+
+ List<FileStoreTable> unsupportedTables =
+ Arrays.asList(
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.SCAN_MODE.key(),
+
CoreOptions.StartupMode.COMPACTED_FULL.toString())),
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), "0")),
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.SCAN_CREATION_TIME_MILLIS.key(), "0")),
+ incrementalTable);
+ for (FileStoreTable unsupportedTable : unsupportedTables) {
+ ReadBuilder unsupportedReadBuilder =
unsupportedTable.newReadBuilder().withTopN(topN);
+ assertThat(unsupportedReadBuilder.newScan().plan().splits())
+ .isNotEmpty()
+ .allMatch(DataSplit.class::isInstance);
+ }
+
+ ReadBuilder readBuilder =
incrementalTable.newReadBuilder().withTopN(topN);
+ TableScan.Plan plan = readBuilder.newScan().plan();
+
+
assertThat(plan.splits()).isNotEmpty().allMatch(DataSplit.class::isInstance);
+ assertThat(readF1(readBuilder, plan))
+ .containsExactlyInAnyOrder(
+ "a100", "a101", "a102", "a103", "a104", "a105",
"a106", "a107", "a108",
+ "a109");
+ }
+
@Test
public void testMixedRowIdOrSkipsGlobalIndexScan() throws Exception {
write(10L);
@@ -195,6 +445,14 @@ public class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
new PredicateBuilder(table.rowType()).equal(1,
BinaryString.fromString("a7"));
table.newReadBuilder().withFilter(predicate).newScan().plan();
+ TopN topN =
+ new TopN(
+ new FieldRef(1, "f1",
table.rowType().getTypeAt(1)),
+ DESCENDING,
+ NULLS_LAST,
+ 1);
+ table.newReadBuilder().withTopN(topN).newScan().plan();
+
PredicateBuilder rowIdBuilder =
new
PredicateBuilder(SpecialFields.rowTypeWithRowId(table.rowType()));
int rowIdIndex = table.rowType().getFieldCount();
@@ -210,6 +468,10 @@ public class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
"INFO Scan table '[^']+' with global index\\. "
+ "searchMode='fast', total=\\d+ ms,
metadata=\\d+ ms, "
+ "lookup=\\d+ ms, coverage=\\d+ ms\\.")
+ .containsPattern(
+ "INFO Scan table '[^']+' with BTree global index
TopN\\. "
+ + "searchMode='fast', topN='[^']+',
total=\\d+ ms, "
+ + "metadata=\\d+ ms, lookup=\\d+ ms,
coverage=\\d+ ms\\.")
.containsPattern(
"INFO Global index lookup table='[^']+',
type='btree', "
+ "fields='\\[f1\\]', lookup=\\d+ ms\\.")
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
index f88993dba8..ba66e09697 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
@@ -20,6 +20,7 @@ package org.apache.paimon.spark.sql
import org.apache.paimon.Snapshot.CommitKind
import org.apache.paimon.errors.ErrorMessages
+import org.apache.paimon.globalindex.IndexedSplit
import org.apache.paimon.spark.PaimonMetrics.RESULTED_TABLE_FILES
import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.spark.read.PaimonSplitScan
@@ -1086,6 +1087,35 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: BTree global index TopN with Spark SQL") {
+ assume(gteqSpark3_3)
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, name STRING) TBLPROPERTIES (
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 'a'), (2, 'c'), (3, 'b'), (4, 'e'), (5,
'd')")
+ sql(
+ "CALL sys.create_global_index(table => 'test.t', index_column =>
'name', " +
+ "index_type => 'btree', options =>
'btree-index.records-per-range=2')")
+
+ val descending = "SELECT id, name FROM t ORDER BY name DESC NULLS LAST
LIMIT 2"
+ val descendingScan = getPaimonScan(descending)
+ assert(descendingScan.pushedTopN.nonEmpty)
+ assert(descendingScan.inputSplits.nonEmpty)
+ assert(descendingScan.inputSplits.forall(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(sql(descending), Seq(Row(4, "e"), Row(5, "d")))
+
+ val ascending = "SELECT id, name FROM t ORDER BY name ASC NULLS LAST
LIMIT 2"
+ val ascendingScan = getPaimonScan(ascending)
+ assert(ascendingScan.pushedTopN.nonEmpty)
+ assert(ascendingScan.inputSplits.nonEmpty)
+ assert(ascendingScan.inputSplits.forall(_.isInstanceOf[IndexedSplit]))
+ checkAnswer(sql(ascending), Seq(Row(1, "a"), Row(3, "b")))
+ }
+ }
+
test("Data Evolution: V1 update table with data-evolution without
condition") {
withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
withTable("t") {
