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 0f8a0e3ccdc9..49958cc3be11 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.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.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 Optional evaluate(@Nullable Predicate predicate) { if (predicate == null) { return Optional.empty(); } + return awaitGlobalIndexResult(visitAsync(predicate)); + } + + public Optional evaluateTopN(TopN topN) { + FieldRef fieldRef = topN.orders().get(0).field(); + int fieldId = rowType.getField(fieldRef.name()).id(); + Collection 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 awaitGlobalIndexResult( + CompletableFuture> 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 807ecb8aa08b..b857fdc1a18e 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.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 @@ default CompletableFuture> visitFullTextSearch throw new UnsupportedOperationException(); } + /** + * Returns row candidates for the given TopN predicate. + * + *

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> visitTopN(TopN topN) { + return CompletableFuture.completedFuture(Optional.empty()); + } + /** Batch search; result {@code i} matches vector {@code i}. */ default CompletableFuture>> 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 7b513a2c3922..23d38d3a9463 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 @@ 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 CompletableFuture>> visitBatchVect }); } + @Override + public CompletableFuture> visitTopN(TopN topN) { + return wrapped.visitTopN(topN).thenApply(this::applyOffset); + } + private Optional applyOffset(Optional 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 5d0195b3fce3..f16c2036fe0a 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 implements GlobalIndexReader { private final SortedFileMetaSelector fileSelector; + private final List files; private final long fallbackScanMaxSize; private final Map readerCache; private final ExecutorService executor; @@ -59,6 +60,7 @@ protected SortedFileGlobalIndexReader( 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 @@ private CompletableFuture> visitSelectedFiles( .thenApply(v -> unionResults(futures)); } + protected CompletableFuture> visitAllFiles( + Function> 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 000000000000..6d4e05b9fb9a --- /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. + * + *

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; + private final Comparator keyComparator; + private final SortValue.SortDirection direction; + private final SortValue.NullOrdering nullOrdering; + private final int limit; + private final RoaringNavigableMap64 results; + + private TopNGlobalIndexResult( + List keyRowIds, + Comparator keyComparator, + SortValue.SortDirection direction, + SortValue.NullOrdering nullOrdering, + int limit) { + this.keyComparator = keyComparator; + this.direction = direction; + this.nullOrdering = nullOrdering; + this.limit = limit; + + List sorted = new ArrayList<>(keyRowIds); + sorted.sort(keyRowIdsComparator()); + this.keyRowIds = Collections.unmodifiableList(mergeAndLimit(sorted)); + this.results = toBitmap(this.keyRowIds); + } + + public static TopNGlobalIndexResult create( + List keyRowIds, + Comparator keyComparator, + SortValue.NullOrdering nullOrdering, + int limit) { + return create( + keyRowIds, keyComparator, SortValue.SortDirection.DESCENDING, nullOrdering, limit); + } + + public static TopNGlobalIndexResult create( + List keyRowIds, + Comparator 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 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 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 mergeAndLimit(List sorted) { + List 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 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) { + RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); + for (KeyRowIds group : keyRowIds) { + for (long rowId : group.rowIds()) { + bitmap.add(rowId); + } + } + return bitmap; + } + + List 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 473eb24d0a84..0e249b5df7db 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 CompletableFuture> visitVectorSearch( }); } + @Override + public CompletableFuture> visitTopN(TopN topN) { + return unionAsync(reader -> reader.visitTopN(topN)); + } + private CompletableFuture> unionAsync( Function>> 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 e9cb833a6814..d8183337bddf 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.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 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 Optional visitBetween(Object from, Object to) { return createResult(() -> rangeQuery(from, to, true, true)); } + public Optional visitTopN(TopN topN) { + List 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 createResult(IOSupplier supplier) { try { return Optional.of(GlobalIndexResult.create(supplier.get())); @@ -362,6 +381,84 @@ private RoaringNavigableMap64 allNonNullRows() throws IOException { return rangeQuery(minKey, maxKey, true, true); } + private TopNGlobalIndexResult topN( + int limit, SortValue.SortDirection direction, SortValue.NullOrdering nullOrdering) + throws IOException { + List 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 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 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 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 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 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 @@ private RoaringNavigableMap64 rangeQuery( } 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 e7e6217e2720..d346d868e7fa 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.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 @@ protected RoaringNavigableMap64 greaterThan(BTreeIndexReader reader, Object lite return bitmap(reader.visitGreaterThan(literal)); } + @Override + public CompletableFuture> 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 caee5884d50f..6c6d0749b154 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 BlockIterator iterator() { 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 000000000000..3ed480e595fc --- /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> { + + 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 788e926dc7e7..21b7a421fac5 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 SstFileIterator createIterator() { 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 @@ private BlockIterator getNextBlock(BlockIterator indexBlockIterator) { 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 BlockIterator readBatch() throws IOException { 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 dcd36c8eaa88..42ccc570e271 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.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.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 @@ void testSingleFieldSequential() { evaluator.close(); } + @Test + void testTopNUsesAggregatedReaderAndReusesPredicateCache() { + RowType rowType = rowType(); + AtomicInteger readersCreated = new AtomicInteger(); + GlobalIndexReader first = + new StubGlobalIndexReader(null) { + @Override + public CompletableFuture> visitTopN(TopN topN) { + return CompletableFuture.completedFuture(Optional.of(resultOf(1, 2))); + } + }; + GlobalIndexReader second = + new StubGlobalIndexReader(null) { + @Override + public CompletableFuture> 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 firstResult = evaluator.evaluateTopN(topN); + Optional 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 000000000000..b5e4e0089572 --- /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 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 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 3c5e9b07c0d6..21a9870a79ea 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 @@ 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 @@ protected GlobalIndexReader prepareDataAndCreateReader() throws Exception { 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 760c5daa9f18..d8321d76845b 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.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.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 @@ private int firstStrictlyGreaterKeyIndex() { return -1; } + @TestTemplate + public void testGlobalTopNCandidatesAcrossFiles() throws Exception { + int limit = 5; + List 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 void testUnorderedIterator() throws Exception { // 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 92e6ae8d2e12..f40b46439903 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 void testUnalignedIterator() throws IOException { 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 void innerTest(boolean aligned) throws IOException { 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 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 000000000000..69b52276c479 --- /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. + * + *

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 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 select(List files, DataField field, TopN topN) { + int limit = topN.limit(); + if (limit == 0) { + return new ArrayList<>(); + } + + BTreeTopNIndexFileSelector selector = new BTreeTopNIndexFileSelector(field, topN); + List selected = new ArrayList<>(); + List 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 6dbbca8e7705..059d66160887 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 InnerTableScan withBucket(int bucket) { @Override public InnerTableScan withTopN(TopN topN) { - batchScan.withTopN(topN); + this.topN = topN; return this; } @@ -166,6 +168,7 @@ public InnerTableScan withMetricRegistry(MetricRegistry metricsRegistry) { @Override public InnerTableScan withLimit(int limit) { + this.pushDownLimit = limit; batchScan.withLimit(limit); return this; } @@ -255,9 +258,16 @@ public List listPartitionEntries() { public Plan plan() { RowRangeIndex rowRangeIndex = this.pushedRowRangeIndex; ScoreGetter scoreGetter = null; + boolean globalIndexTopNCandidatesFound = false; if (rowRangeIndex == null) { - Optional indexResult = evalGlobalIndex(); + Optional 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 Plan plan() { } } + if (!globalIndexTopNCandidatesFound && topN != null) { + batchScan.withTopN(topN); + } + if (rowRangeIndex == null) { return batchScan.plan(); } @@ -328,6 +342,73 @@ private Optional evalGlobalIndex() { } } + private Optional evalGlobalIndexTopN() { + CoreOptions options = table.coreOptions(); + PartitionPredicate partitionFilter = + batchScan.snapshotReader().manifestsReader().partitionFilter(); + long totalStart = System.nanoTime(); + Optional 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 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 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 7dbdb06853d0..498130a11dd3 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 @@ 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.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 @@ /** 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 @@ private DataEvolutionGlobalIndexScanner( FileIO fileIO, IndexPathFactory indexPathFactory, Collection 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 coverageIndexFiles, + Collection indexFiles) { this.table = table; this.options = options; this.rowType = rowType; @@ -97,7 +123,7 @@ private DataEvolutionGlobalIndexScanner( table, snapshot, partitionFilter, - indexFiles, + coverageIndexFiles, table.coreOptions().scalarIndexSearchMode()); GlobalIndexFileReader indexFileReader = meta -> fileIO.newInputStream(meta.filePath()); Map indexMetas = new HashMap<>(); @@ -128,7 +154,6 @@ private DataEvolutionGlobalIndexScanner( } group.addFile(indexType, range, indexFile); } - IntFunction> readersFunction = fId -> { List groups = new ArrayList<>(); @@ -244,6 +269,65 @@ public static Optional create( indexFiles)); } + /** + * Creates a scanner for a single-column TopN backed by a primary BTree index. + * + *

Indexes carrying the ordered field only as an extra field are not ordered by that field + * and cannot serve this scan. + */ + public static Optional 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 indexFiles = + table.store().newIndexFileHandler() + .scan(snapshot, topNIndexFileFilter(partitionFilter, fieldId)).stream() + .map(IndexManifestEntry::indexFile) + .collect(Collectors.toList()); + if (indexFiles.isEmpty()) { + return Optional.empty(); + } + List 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 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 indexFileFilter( FileStoreTable table, @Nullable PartitionPredicate partitionFilter, @@ -288,6 +372,20 @@ public Optional scan(Predicate predicate) { return globalIndexEvaluator.evaluate(predicate); } + public Optional 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 GlobalIndexResult unindexedRows(Predicate predicate) { 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 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 000000000000..044f2863458d --- /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 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 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 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 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 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 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 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 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 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 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 select( + List files, SortValue.NullOrdering nullOrdering, int limit) { + return select(files, DESCENDING, nullOrdering, limit); + } + + private List select( + List 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 fileNames(List 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 676a1c305e9e..375cac3989ff 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.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.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 void testBTreeGlobalIndexWithCoreScan() throws Exception { 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 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 void testGlobalIndexDiagnosticLogs() throws Exception { 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 void testGlobalIndexDiagnosticLogs() throws Exception { "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 f88993dba846..ba66e09697ba 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") {