This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit c5d59d0ff0c5307f627bed333c23f5e992ec7114 Author: Jingsong Lee <[email protected]> AuthorDate: Sat Sep 19 23:21:29 2026 +0800 [core] Add adaptive BTree posting list encoding (#9981) --- .../paimon/globalindex/btree/BTreeFileFooter.java | 13 +- .../globalindex/btree/BTreeGlobalIndexer.java | 3 +- .../globalindex/btree/BTreeIndexOptions.java | 9 + .../paimon/globalindex/btree/BTreeIndexReader.java | 56 +++- .../paimon/globalindex/btree/BTreeIndexWriter.java | 56 +++- .../paimon/globalindex/btree/BTreePostingList.java | 307 +++++++++++++++++++++ .../apache/paimon/utils/RoaringNavigableMap64.java | 25 +- .../globalindex/btree/AbstractIndexReaderTest.java | 7 +- .../globalindex/btree/BTreeIndexOptionsTest.java | 6 + .../globalindex/btree/BTreeIndexReaderTest.java | 57 +++- .../globalindex/btree/BTreePostingListTest.java | 232 ++++++++++++++++ .../paimon/utils/RoaringNavigableMap64Test.java | 26 ++ 12 files changed, 763 insertions(+), 34 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileFooter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileFooter.java index 19994921c3..c58d7950df 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileFooter.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileFooter.java @@ -32,7 +32,16 @@ import static org.apache.paimon.utils.Preconditions.checkArgument; public class BTreeFileFooter { public static final int MAGIC_NUMBER = 0x50425449; - public static final int CURRENT_VERSION = 1; + public static final int VERSION_1 = 1; + public static final int VERSION_2 = 2; + + // Keep writing version 1 until all readers in the ecosystem support version 2. + public static final int DEFAULT_WRITE_VERSION = VERSION_1; + public static final int MAX_SUPPORTED_VERSION = VERSION_2; + + /** The default version written by constructors which do not specify a version. */ + public static final int CURRENT_VERSION = DEFAULT_WRITE_VERSION; + public static final int ENCODED_LENGTH = 52; private final int version; @@ -44,7 +53,7 @@ public class BTreeFileFooter { @Nullable BloomFilterHandle bloomFilterHandle, BlockHandle indexBlockHandle, BlockHandle nullBitmapHandle) { - this(CURRENT_VERSION, bloomFilterHandle, indexBlockHandle, nullBitmapHandle); + this(DEFAULT_WRITE_VERSION, bloomFilterHandle, indexBlockHandle, nullBitmapHandle); } public BTreeFileFooter( diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java index 0b4c5b4763..7196e18a77 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java @@ -108,7 +108,8 @@ public class BTreeGlobalIndexer implements SortedGlobalIndexer { keySerializer, (int) blockSize, bloomFilterBuilder, - BlockCompressionFactory.create(compressOptions)); + BlockCompressionFactory.create(compressOptions), + options.get(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION)); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java index e06704c561..48c1068b2c 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java @@ -25,6 +25,15 @@ import org.apache.paimon.options.MemorySize; /** Options for BTree index. */ public class BTreeIndexOptions { + public static final ConfigOption<Integer> BTREE_INDEX_FILE_VERSION = + ConfigOptions.key("btree-index.file-version") + .intType() + .defaultValue(BTreeFileFooter.DEFAULT_WRITE_VERSION) + .withDescription( + "The BTree index file version to write. Version 1 is the default for " + + "reader compatibility; version 2 enables adaptive posting " + + "list encoding and requires all readers to support it."); + public static final ConfigOption<String> BTREE_INDEX_COMPRESSION = ConfigOptions.key("btree-index.compression") .stringType() 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 b42224106d..fc2c480d71 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 @@ -67,6 +67,7 @@ public class BTreeIndexReader implements Closeable { private final SstFileReader reader; private final KeySerializer keySerializer; private final Comparator<Object> comparator; + private final int fileVersion; private final LazyField<RoaringNavigableMap64> nullBitmap; private final Object minKey; private final Object maxKey; @@ -161,6 +162,12 @@ public class BTreeIndexReader implements Closeable { Path filePath = globalIndexIOMeta.filePath(); BlockCache blockCache = new BlockCache(filePath, input, cacheManager); BTreeFileFooter footer = readFooter(blockCache, fileSize); + this.fileVersion = footer.getVersion(); + Preconditions.checkArgument( + fileVersion >= BTreeFileFooter.VERSION_1 + && fileVersion <= BTreeFileFooter.MAX_SUPPORTED_VERSION, + "Unsupported BTree index version: %s", + fileVersion); // prepare nullBitmap and SstFileReader this.nullBitmap = @@ -408,9 +415,7 @@ public class BTreeIndexReader implements Closeable { RoaringNavigableMap64 result = new RoaringNavigableMap64(); byte[] rowIds = reader.lookup(keySerializer.serialize(key)); if (rowIds != null) { - for (long rowId : deserializeRowIds(MemorySlice.wrap(rowIds))) { - result.add(rowId); - } + addRowIdsTo(MemorySlice.wrap(rowIds), result); } return result; } @@ -524,28 +529,49 @@ public class BTreeIndexReader implements Closeable { return result; } - for (long rowId : deserializeRowIds(entry.getValue())) { - result.add(rowId); - } + addRowIdsTo(entry.getValue(), result); } } return result; } - private long[] deserializeRowIds(MemorySlice slice) { + private long[] deserializeRowIds(MemorySlice slice) throws IOException { return deserializeRowIds(slice, Integer.MAX_VALUE); } - static long[] deserializeRowIds(MemorySlice slice, int maxRowIds) { + private long[] deserializeRowIds(MemorySlice slice, int maxRowIds) throws IOException { + return fileVersion == BTreeFileFooter.VERSION_1 + ? deserializeVersion1RowIds(slice, maxRowIds) + : BTreePostingList.deserialize(slice, maxRowIds); + } + + private void addRowIdsTo(MemorySlice slice, RoaringNavigableMap64 target) throws IOException { + if (fileVersion == BTreeFileFooter.VERSION_1) { + MemorySliceInput input = slice.toInput(); + int count = readVersion1Count(input); + for (int i = 0; i < count; i++) { + target.add(input.readVarLenLong()); + } + } else { + BTreePostingList.addTo(slice, target); + } + } + + static long[] deserializeVersion1RowIds(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"); - int resultLength = Math.min(length, maxRowIds); - long[] ids = new long[resultLength]; + MemorySliceInput input = slice.toInput(); + int count = readVersion1Count(input); + int resultLength = Math.min(count, maxRowIds); + long[] result = new long[resultLength]; for (int i = 0; i < resultLength; i++) { - ids[i] = sliceInput.readVarLenLong(); + result[i] = input.readVarLenLong(); } - return ids; + return result; + } + + private static int readVersion1Count(MemorySliceInput input) { + int count = input.readVarLenInt(); + Preconditions.checkState(count > 0, "Invalid BTree row id count: %s", count); + return count; } } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java index e1916d11a4..b8d6f62bea 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java @@ -32,12 +32,12 @@ import org.apache.paimon.sst.BloomFilterHandle; import org.apache.paimon.sst.SstFileWriter; import org.apache.paimon.utils.BloomFilter; import org.apache.paimon.utils.LazyField; +import org.apache.paimon.utils.LongArrayList; import org.apache.paimon.utils.RoaringNavigableMap64; import javax.annotation.Nullable; import java.io.IOException; -import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; import java.util.List; @@ -45,8 +45,9 @@ import java.util.zip.CRC32; /** * The {@link GlobalIndexSingleColumnWriter} implementation for BTree index. Note that users must - * keep written keys monotonically incremental. All null keys are stored in a separate bitmap, which - * will be serialized and appended to the file end on close. The layout is as below: + * keep written keys monotonically incremental and row ids for the same non-null key strictly + * increasing. All null keys are stored in a separate bitmap, which will be serialized and appended + * to the file end on close. The layout is as below: * * <pre> * +-----------------------------------+------+ @@ -77,7 +78,8 @@ public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter { private final SstFileWriter writer; private final KeySerializer keySerializer; private final Comparator<Object> comparator; - private final List<Long> currentRowIds = new ArrayList<>(); + private final int fileVersion; + private final LongArrayList currentRowIds = new LongArrayList(32); private final LazyField<RoaringNavigableMap64> nullBitmap = new LazyField<>(RoaringNavigableMap64::new); @@ -101,10 +103,32 @@ public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter { @Nullable BloomFilter.Builder bloomFilterBuilder, BlockCompressionFactory compressionFactory) throws IOException { + this( + indexFileWriter, + keySerializer, + blockSize, + bloomFilterBuilder, + compressionFactory, + BTreeFileFooter.DEFAULT_WRITE_VERSION); + } + + BTreeIndexWriter( + GlobalIndexFileWriter indexFileWriter, + KeySerializer keySerializer, + int blockSize, + @Nullable BloomFilter.Builder bloomFilterBuilder, + BlockCompressionFactory compressionFactory, + int fileVersion) + throws IOException { + if (fileVersion < BTreeFileFooter.VERSION_1 + || fileVersion > BTreeFileFooter.MAX_SUPPORTED_VERSION) { + throw new IllegalArgumentException("Unsupported BTree index version: " + fileVersion); + } this.fileName = indexFileWriter.newFileName(BTreeGlobalIndexerFactory.IDENTIFIER); this.out = indexFileWriter.newOutputStream(this.fileName); this.keySerializer = keySerializer; this.comparator = keySerializer.createComparator(); + this.fileVersion = fileVersion; this.writer = new SstFileWriter(out, blockSize, bloomFilterBuilder, compressionFactory); } @@ -137,15 +161,22 @@ public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter { return; } - // serialize row id list - MemorySliceOutput sliceOutput = new MemorySliceOutput(currentRowIds.size() * 9 + 5); - sliceOutput.writeVarLenInt(currentRowIds.size()); - for (long currentRowId : currentRowIds) { - sliceOutput.writeVarLenLong(currentRowId); - } + byte[] postingList = + fileVersion == BTreeFileFooter.VERSION_1 + ? serializeVersion1RowIds(currentRowIds) + : BTreePostingList.serialize(currentRowIds); currentRowIds.clear(); - writer.put(keySerializer.serialize(lastKey), sliceOutput.toSlice().copyBytes()); + writer.put(keySerializer.serialize(lastKey), postingList); + } + + private static byte[] serializeVersion1RowIds(LongArrayList rowIds) { + MemorySliceOutput output = new MemorySliceOutput(rowIds.size() * 9 + 5); + output.writeVarLenInt(rowIds.size()); + for (int i = 0; i < rowIds.size(); i++) { + output.writeVarLenLong(rowIds.get(i)); + } + return output.toSlice().copyBytes(); } @Override @@ -168,7 +199,8 @@ public class BTreeIndexWriter implements GlobalIndexSingleColumnWriter { // write footer BTreeFileFooter footer = - new BTreeFileFooter(bloomFilterHandle, indexBlockHandle, nullBitmapHandle); + new BTreeFileFooter( + fileVersion, bloomFilterHandle, indexBlockHandle, nullBitmapHandle); MemorySlice footerEncoding = BTreeFileFooter.writeFooter(footer); writer.writeSlice(footerEncoding); diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreePostingList.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreePostingList.java new file mode 100644 index 0000000000..8be3ba0492 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreePostingList.java @@ -0,0 +1,307 @@ +/* + * 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.btree; + +import org.apache.paimon.memory.MemorySlice; +import org.apache.paimon.memory.MemorySliceInput; +import org.apache.paimon.memory.MemorySliceOutput; +import org.apache.paimon.utils.LongArrayList; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import java.io.IOException; + +import static org.apache.paimon.utils.Preconditions.checkArgument; +import static org.apache.paimon.utils.Preconditions.checkState; + +/** + * Self-describing adaptive encoding for the row ids associated with one BTree key. + * + * <p>Singletons have a dedicated representation. Larger postings use the smaller serialized form of + * a delta list and a Roaring bitmap. A conservative lower bound avoids constructing the Roaring + * candidate when it cannot be smaller. + */ +final class BTreePostingList { + + static final int SINGLE = 0; + static final int DELTA_LIST = 1; + static final int ROARING = 2; + + private static final int MIN_ROARING_RANGE_LENGTH = 64; + + private BTreePostingList() {} + + static byte[] serialize(LongArrayList rowIds) throws IOException { + checkArgument(!rowIds.isEmpty(), "Cannot serialize an empty BTree posting list."); + if (rowIds.size() == 1) { + long rowId = rowIds.get(0); + checkNonNegative(rowId); + MemorySliceOutput output = new MemorySliceOutput(1 + varLenSize(rowId)); + output.writeByte(SINGLE); + output.writeVarLenLong(rowId); + return output.toSlice().getHeapMemory(); + } + + EncodingSizes sizes = estimateEncodingSizes(rowIds); + if (sizes.roaringLowerBound >= sizes.deltaList) { + return serializeDeltaList(rowIds, sizes.deltaList); + } + + byte[] roaring = serializeRoaring(rowIds); + return roaring.length < sizes.deltaList + ? roaring + : serializeDeltaList(rowIds, sizes.deltaList); + } + + static void addTo(MemorySlice slice, RoaringNavigableMap64 target) throws IOException { + MemorySliceInput input = slice.toInput(); + int type = input.readUnsignedByte(); + switch (type) { + case SINGLE: + target.add(input.readVarLenLong()); + return; + case DELTA_LIST: + addDeltaList(input, target); + return; + case ROARING: + target.or(readRoaring(input)); + return; + default: + throw new IllegalStateException("Unknown BTree posting list type: " + type); + } + } + + static long[] deserialize(MemorySlice slice, int maxRowIds) throws IOException { + checkArgument(maxRowIds >= 0, "Max row id count must not be negative."); + MemorySliceInput input = slice.toInput(); + int type = input.readUnsignedByte(); + switch (type) { + case SINGLE: + return maxRowIds == 0 ? new long[0] : new long[] {input.readVarLenLong()}; + case DELTA_LIST: + return deserializeDeltaList(input, maxRowIds); + case ROARING: + return first(readRoaring(input), maxRowIds); + default: + throw new IllegalStateException("Unknown BTree posting list type: " + type); + } + } + + private static EncodingSizes estimateEncodingSizes(LongArrayList rowIds) { + int deltaListSize = 1 + varLenSize(rowIds.size()); + long previous = rowIds.get(0); + checkNonNegative(previous); + deltaListSize += varLenSize(previous); + + long roaringLowerBound = 1L + Long.BYTES + Integer.BYTES + Integer.BYTES; + long currentHigh = previous >>> 32; + long currentContainer = previous >>> 16; + int containerCardinality = 1; + int containerRuns = 1; + for (int i = 1; i < rowIds.size(); i++) { + long current = rowIds.get(i); + checkIncreasing(current, previous); + deltaListSize += varLenSize(current - previous); + + long high = current >>> 32; + long container = current >>> 16; + if (container != currentContainer) { + roaringLowerBound += containerLowerBound(containerCardinality, containerRuns); + if (high != currentHigh) { + // High key plus the nested Roaring bitmap cookie. + roaringLowerBound += Integer.BYTES + Integer.BYTES; + currentHigh = high; + } + currentContainer = container; + containerCardinality = 1; + containerRuns = 1; + } else { + containerCardinality++; + if (previous == Long.MAX_VALUE || current != previous + 1) { + containerRuns++; + } + } + previous = current; + } + roaringLowerBound += containerLowerBound(containerCardinality, containerRuns); + return new EncodingSizes(deltaListSize, roaringLowerBound); + } + + private static byte[] serializeDeltaList(LongArrayList rowIds, int serializedSize) { + MemorySliceOutput output = new MemorySliceOutput(serializedSize); + output.writeByte(DELTA_LIST); + output.writeVarLenInt(rowIds.size()); + long previous = rowIds.get(0); + output.writeVarLenLong(previous); + for (int i = 1; i < rowIds.size(); i++) { + long current = rowIds.get(i); + output.writeVarLenLong(current - previous); + previous = current; + } + checkState( + output.size() == serializedSize, + "Unexpected delta BTree posting list size: %s instead of %s", + output.size(), + serializedSize); + return output.toSlice().getHeapMemory(); + } + + private static long containerLowerBound(int cardinality, int runs) { + long arrayBytes = 2L * cardinality; + long bitmapBytes = 8192; + long runBytes = 2L + 4L * runs; + // Every container also needs a two-byte key and a two-byte cardinality. + return Integer.BYTES + Math.min(Math.min(arrayBytes, bitmapBytes), runBytes); + } + + private static byte[] serializeRoaring(LongArrayList rowIds) throws IOException { + RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); + long rangeStart = rowIds.get(0); + long previous = rangeStart; + for (int i = 0; i < rowIds.size(); i++) { + long current = rowIds.get(i); + if (i > 0 && current != previous + 1) { + addRange(bitmap, rangeStart, previous); + rangeStart = current; + } + previous = current; + } + addRange(bitmap, rangeStart, previous); + + byte[] bytes = bitmap.serialize(); + byte[] result = new byte[bytes.length + 1]; + result[0] = ROARING; + System.arraycopy(bytes, 0, result, 1, bytes.length); + return result; + } + + private static void addRange(RoaringNavigableMap64 bitmap, long from, long to) { + if (to - from >= MIN_ROARING_RANGE_LENGTH - 1) { + if (to == Long.MAX_VALUE) { + bitmap.addRange(new Range(from, to - 1)); + bitmap.add(to); + } else { + bitmap.addRange(new Range(from, to)); + } + return; + } + + for (long rowId = from; ; rowId++) { + bitmap.add(rowId); + if (rowId == to) { + return; + } + } + } + + private static void addDeltaList(MemorySliceInput input, RoaringNavigableMap64 target) { + int count = readDeltaCount(input); + long rowId = input.readVarLenLong(); + target.add(rowId); + for (int i = 1; i < count; i++) { + rowId += readPositiveDelta(input); + target.add(rowId); + } + } + + private static long[] deserializeDeltaList(MemorySliceInput input, int maxRowIds) { + int count = readDeltaCount(input); + int resultLength = Math.min(count, maxRowIds); + if (resultLength == 0) { + return new long[0]; + } + + long[] result = new long[resultLength]; + result[0] = input.readVarLenLong(); + for (int i = 1; i < resultLength; i++) { + result[i] = result[i - 1] + readPositiveDelta(input); + } + return result; + } + + private static int readDeltaCount(MemorySliceInput input) { + int count = input.readVarLenInt(); + if (count <= 1) { + throw new IllegalStateException("Invalid delta BTree posting list length: " + count); + } + return count; + } + + private static long readPositiveDelta(MemorySliceInput input) { + long delta = input.readVarLenLong(); + if (delta <= 0) { + throw new IllegalStateException("Invalid non-positive BTree row id delta: " + delta); + } + return delta; + } + + private static RoaringNavigableMap64 readRoaring(MemorySliceInput input) throws IOException { + RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); + MemorySlice payload = input.readSlice(input.available()); + byte[] heapMemory = payload.getHeapMemory(); + if (heapMemory == null) { + bitmap.deserialize(payload.copyBytes()); + } else { + bitmap.deserialize(heapMemory, payload.offset(), payload.length()); + } + checkState(!bitmap.isEmpty(), "Invalid empty Roaring BTree posting list."); + return bitmap; + } + + private static long[] first(RoaringNavigableMap64 bitmap, int maxRowIds) { + return bitmap.toArray(maxRowIds); + } + + private static int varLenSize(long value) { + int size = 1; + while ((value & ~0x7FL) != 0) { + value >>>= 7; + size++; + } + return size; + } + + private static void checkNonNegative(long rowId) { + if (rowId < 0) { + throw new IllegalArgumentException( + "BTree row id must be non-negative, but was " + rowId); + } + } + + private static void checkIncreasing(long current, long previous) { + if (current <= previous) { + throw new IllegalArgumentException( + "BTree row ids must be strictly increasing, but found " + + current + + " after " + + previous + + '.'); + } + } + + private static class EncodingSizes { + private final int deltaList; + private final long roaringLowerBound; + + private EncodingSizes(int deltaList, long roaringLowerBound) { + this.deltaList = deltaList; + this.roaringLowerBound = roaringLowerBound; + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/RoaringNavigableMap64.java b/paimon-common/src/main/java/org/apache/paimon/utils/RoaringNavigableMap64.java index e82382258a..006c92a46c 100644 --- a/paimon-common/src/main/java/org/apache/paimon/utils/RoaringNavigableMap64.java +++ b/paimon-common/src/main/java/org/apache/paimon/utils/RoaringNavigableMap64.java @@ -109,6 +109,18 @@ public class RoaringNavigableMap64 implements Iterable<Long>, Serializable { return roaring64NavigableMap.iterator(); } + /** Returns at most the first {@code maxValues} values without boxing them. */ + public long[] toArray(int maxValues) { + Preconditions.checkArgument(maxValues >= 0, "Max value count must not be negative."); + int resultLength = (int) Math.min(getLongCardinality(), (long) maxValues); + long[] result = new long[resultLength]; + LongIterator iterator = roaring64NavigableMap.getLongIterator(); + for (int i = 0; i < resultLength; i++) { + result[i] = iterator.next(); + } + return result; + } + public static RoaringNavigableMap64 and(RoaringNavigableMap64 x1, RoaringNavigableMap64 x2) { Roaring64NavigableMap result = new Roaring64NavigableMap(); result.or(x1.roaring64NavigableMap); @@ -149,7 +161,18 @@ public class RoaringNavigableMap64 implements Iterable<Long>, Serializable { } public void deserialize(byte[] rbmBytes) throws IOException { - try (ByteArrayInputStream bis = new ByteArrayInputStream(rbmBytes); + deserialize(rbmBytes, 0, rbmBytes.length); + } + + /** Deserializes a bitmap from a region of the given byte array without copying that region. */ + public void deserialize(byte[] rbmBytes, int offset, int length) throws IOException { + Preconditions.checkArgument( + offset >= 0 && length >= 0 && offset <= rbmBytes.length - length, + "Invalid bitmap byte range [%s, %s) for array length %s.", + offset, + (long) offset + length, + rbmBytes.length); + try (ByteArrayInputStream bis = new ByteArrayInputStream(rbmBytes, offset, length); DataInputStream dis = new DataInputStream(bis)) { roaring64NavigableMap.deserializePortable(dis); } diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java index bbb0e2d7e0..c00f26bff5 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java @@ -334,7 +334,12 @@ public abstract class AbstractIndexReaderTest { protected abstract GlobalIndexReader prepareDataAndCreateReader() throws Exception; protected GlobalIndexIOMeta writeData(List<Pair<Object, Long>> data) throws IOException { - GlobalIndexSingleColumnWriter indexWriter = globalIndexer.createWriter(fileWriter); + return writeData(data, globalIndexer.createWriter(fileWriter)); + } + + protected GlobalIndexIOMeta writeData( + List<Pair<Object, Long>> data, GlobalIndexSingleColumnWriter indexWriter) + throws IOException { for (Pair<Object, Long> pair : data) { indexWriter.write(pair.getKey(), pair.getValue()); } diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java index 72ceec750b..de0ad667c7 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexOptionsTest.java @@ -25,6 +25,12 @@ import static org.assertj.core.api.Assertions.assertThat; /** Tests for {@link BTreeIndexOptions}. */ class BTreeIndexOptionsTest { + @Test + void testDefaultFileVersionRemainsVersion1() { + assertThat(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION.defaultValue()) + .isEqualTo(BTreeFileFooter.VERSION_1); + } + @Test void testDefaultRecordsPerRange() { assertThat(BTreeIndexOptions.BTREE_INDEX_RECORDS_PER_RANGE.defaultValue()) 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 b6e5eed591..7c6dd5b098 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 @@ -18,13 +18,16 @@ package org.apache.paimon.globalindex.btree; +import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.memory.MemorySlice; 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.apache.paimon.utils.IOUtils; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; @@ -134,16 +137,66 @@ public class BTreeIndexReaderTest extends AbstractIndexReaderTest { } @TestTemplate - public void testTopNOnlyDeserializesRemainingRowIds() { + public void testVersion1TopNOnlyDeserializesRemainingRowIds() throws Exception { MemorySliceOutput output = new MemorySliceOutput(16); output.writeVarLenInt(3); output.writeVarLenLong(10); output.writeVarLenLong(20); - assertThat(BTreeIndexReader.deserializeRowIds(output.toSlice(), 2)) + assertThat(BTreeIndexReader.deserializeVersion1RowIds(output.toSlice(), 2)) .containsExactly(10L, 20L); } + @TestTemplate + public void testReadsVersion1File() throws Exception { + assertReadsConfiguredFileVersion(BTreeFileFooter.VERSION_1); + } + + @TestTemplate + public void testReadsVersion2File() throws Exception { + options.set(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION, BTreeFileFooter.VERSION_2); + options.set(BTreeIndexOptions.BTREE_INDEX_COMPRESSION, "lz4"); + assertReadsConfiguredFileVersion(BTreeFileFooter.VERSION_2); + } + + private void assertReadsConfiguredFileVersion(int expectedFileVersion) throws Exception { + GlobalIndexIOMeta written = writeData(data); + assertFileVersion(written, expectedFileVersion); + + FieldRef ref = new FieldRef(1, "testField", dataType); + Object literal = data.get(dataNum / 2).getKey(); + Object from = data.get(dataNum / 3).getKey(); + Object to = data.get(dataNum * 2 / 3).getKey(); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, + Collections.singletonList(written), + dataNum, + newDirectExecutorService())) { + assertResult( + reader.visitEqual(ref, literal).join().get(), + filter(value -> comparator.compare(value, literal) == 0)); + assertResult( + reader.visitBetween(ref, from, to).join().get(), + filter( + value -> + comparator.compare(value, from) >= 0 + && comparator.compare(value, to) <= 0)); + } + } + + private void assertFileVersion(GlobalIndexIOMeta written, int expectedFileVersion) + throws Exception { + byte[] footerBytes = new byte[BTreeFileFooter.ENCODED_LENGTH]; + try (SeekableInputStream input = fileReader.getInputStream(written)) { + input.seek(written.fileSize() - BTreeFileFooter.ENCODED_LENGTH); + IOUtils.readFully(input, footerBytes); + } + assertThat(BTreeFileFooter.readFooter(MemorySlice.wrap(footerBytes).toInput()).getVersion()) + .isEqualTo(expectedFileVersion); + } + private Object[] valuesByRowId() { Object[] values = new Object[dataNum]; data.forEach(pair -> values[pair.getValue().intValue()] = pair.getKey()); diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreePostingListTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreePostingListTest.java new file mode 100644 index 0000000000..1abaabc1a0 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreePostingListTest.java @@ -0,0 +1,232 @@ +/* + * 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.btree; + +import org.apache.paimon.memory.MemorySlice; +import org.apache.paimon.memory.MemorySliceOutput; +import org.apache.paimon.utils.LongArrayList; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link BTreePostingList}. */ +public class BTreePostingListTest { + + @Test + public void testChoosesSingleton() throws Exception { + assertEncodingAndRoundTrip(rowIds(123), BTreePostingList.SINGLE); + } + + @Test + public void testChoosesRoaringForContiguousRows() throws Exception { + assertEncodingAndRoundTrip(contiguous(0, 128), BTreePostingList.ROARING); + } + + @Test + public void testChoosesDeltaListForSparseRows() throws Exception { + LongArrayList rowIds = new LongArrayList(256); + for (int i = 0; i < 256; i++) { + rowIds.add(i * 100L); + } + + assertEncodingAndRoundTrip(rowIds, BTreePostingList.DELTA_LIST); + } + + @Test + public void testChoosesRoaringForLargeContiguousRows() throws Exception { + assertEncodingAndRoundTrip(contiguous(1L << 33, 128), BTreePostingList.ROARING); + } + + @Test + public void testChoosesRoaringForRowsEndingAtMaxValue() throws Exception { + assertEncodingAndRoundTrip(contiguous(Long.MAX_VALUE - 127, 128), BTreePostingList.ROARING); + } + + @Test + public void testChoosesSmallestSerializedEncoding() throws Exception { + assertSmallest(contiguous(0, 10_000)); + + LongArrayList sparse = new LongArrayList(10_000); + for (int i = 0; i < 10_000; i++) { + sparse.add(i * 100L); + } + assertSmallest(sparse); + + LongArrayList multipleHighBuckets = new LongArrayList(4); + multipleHighBuckets.add(0); + multipleHighBuckets.add(1); + multipleHighBuckets.add(1L << 32); + multipleHighBuckets.add((1L << 32) + 1); + assertSmallest(multipleHighBuckets); + } + + @Test + public void testAllAdaptiveEncodingsHonorRowLimit() throws Exception { + assertThat(first(contiguous(0, 1), 0)).isEmpty(); + assertThat(first(contiguous(0, 10), 2)).containsExactly(0L, 1L); + + LongArrayList sparse = new LongArrayList(10); + for (int i = 0; i < 10; i++) { + sparse.add(i * 100L); + } + assertThat(first(sparse, 2)).containsExactly(0L, 100L); + assertThat(first(contiguous(1L << 33, 128), 2)).containsExactly(1L << 33, (1L << 33) + 1); + } + + @Test + public void testRejectsUnsortedRows() { + assertThatThrownBy(() -> BTreePostingList.serialize(rowIds(1, 3, 2))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("strictly increasing"); + + assertThatThrownBy(() -> BTreePostingList.serialize(rowIds(1, 1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("strictly increasing"); + } + + @Test + public void testRejectsUnknownEncoding() { + assertThatThrownBy( + () -> + BTreePostingList.deserialize( + MemorySlice.wrap(new byte[] {99}), Integer.MAX_VALUE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Unknown BTree posting list type"); + } + + @Test + public void testRejectsInvalidDeltaList() { + MemorySliceOutput invalidCount = new MemorySliceOutput(3); + invalidCount.writeByte(BTreePostingList.DELTA_LIST); + invalidCount.writeVarLenInt(1); + invalidCount.writeVarLenLong(0); + assertThatThrownBy( + () -> + BTreePostingList.deserialize( + invalidCount.toSlice(), Integer.MAX_VALUE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Invalid delta BTree posting list length"); + + MemorySliceOutput zeroDelta = new MemorySliceOutput(4); + zeroDelta.writeByte(BTreePostingList.DELTA_LIST); + zeroDelta.writeVarLenInt(2); + zeroDelta.writeVarLenLong(1); + zeroDelta.writeVarLenLong(0); + assertThatThrownBy( + () -> BTreePostingList.deserialize(zeroDelta.toSlice(), Integer.MAX_VALUE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Invalid non-positive BTree row id delta"); + } + + @Test + public void testRejectsEmptyRoaring() throws Exception { + byte[] emptyBitmap = new RoaringNavigableMap64().serialize(); + MemorySliceOutput output = new MemorySliceOutput(emptyBitmap.length + 1); + output.writeByte(BTreePostingList.ROARING); + output.writeBytes(emptyBitmap); + + assertThatThrownBy(() -> BTreePostingList.deserialize(output.toSlice(), Integer.MAX_VALUE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("Invalid empty Roaring BTree posting list"); + } + + private static void assertEncodingAndRoundTrip(LongArrayList rowIds, int expectedEncoding) + throws Exception { + byte[] serialized = BTreePostingList.serialize(rowIds); + assertThat(serialized[0]).isEqualTo((byte) expectedEncoding); + MemorySlice paddedSlice = paddedSlice(serialized); + assertThat(BTreePostingList.deserialize(paddedSlice, Integer.MAX_VALUE)) + .containsExactly(rowIds.toArray()); + + RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); + BTreePostingList.addTo(paddedSlice, bitmap); + List<Long> actual = new ArrayList<>(); + bitmap.iterator().forEachRemaining(actual::add); + assertThat(actual).containsExactlyElementsOf(asList(rowIds)); + } + + private static MemorySlice paddedSlice(byte[] serialized) { + byte[] framed = new byte[serialized.length + 4]; + System.arraycopy(serialized, 0, framed, 2, serialized.length); + return MemorySlice.wrap(framed).slice(2, serialized.length); + } + + private static void assertSmallest(LongArrayList rowIds) throws Exception { + byte[] adaptive = BTreePostingList.serialize(rowIds); + assertThat(adaptive.length) + .isEqualTo(Math.min(deltaSerializedSize(rowIds), roaringSerializedSize(rowIds))); + } + + private static int deltaSerializedSize(LongArrayList rowIds) { + MemorySliceOutput output = new MemorySliceOutput(rowIds.size() + 10); + output.writeByte(BTreePostingList.DELTA_LIST); + output.writeVarLenInt(rowIds.size()); + long previous = rowIds.get(0); + output.writeVarLenLong(previous); + for (int i = 1; i < rowIds.size(); i++) { + long current = rowIds.get(i); + output.writeVarLenLong(current - previous); + previous = current; + } + return output.size(); + } + + private static int roaringSerializedSize(LongArrayList rowIds) throws Exception { + RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); + for (int i = 0; i < rowIds.size(); i++) { + bitmap.add(rowIds.get(i)); + } + return 1 + bitmap.serialize().length; + } + + private static long[] first(LongArrayList rowIds, int maxRowIds) throws Exception { + return BTreePostingList.deserialize( + MemorySlice.wrap(BTreePostingList.serialize(rowIds)), maxRowIds); + } + + private static LongArrayList contiguous(long first, int count) { + LongArrayList rowIds = new LongArrayList(count); + for (int i = 0; i < count; i++) { + rowIds.add(first + i); + } + return rowIds; + } + + private static LongArrayList rowIds(long... values) { + LongArrayList rowIds = new LongArrayList(values.length); + for (long value : values) { + rowIds.add(value); + } + return rowIds; + } + + private static List<Long> asList(LongArrayList rowIds) { + List<Long> result = new ArrayList<>(rowIds.size()); + for (int i = 0; i < rowIds.size(); i++) { + result.add(rowIds.get(i)); + } + return result; + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/utils/RoaringNavigableMap64Test.java b/paimon-common/src/test/java/org/apache/paimon/utils/RoaringNavigableMap64Test.java index 7458247c49..a51262c261 100644 --- a/paimon-common/src/test/java/org/apache/paimon/utils/RoaringNavigableMap64Test.java +++ b/paimon-common/src/test/java/org/apache/paimon/utils/RoaringNavigableMap64Test.java @@ -24,6 +24,7 @@ import java.util.ArrayList; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link RoaringNavigableMap64}. */ public class RoaringNavigableMap64Test { @@ -125,6 +126,31 @@ public class RoaringNavigableMap64Test { assertThat(values.get(100)).isEqualTo(end); } + @Test + public void testToArrayWithLimit() { + RoaringNavigableMap64 bitmap = RoaringNavigableMap64.bitmapOf(1, 3, 5); + + assertThat(bitmap.toArray(0)).isEmpty(); + assertThat(bitmap.toArray(2)).containsExactly(1, 3); + assertThat(bitmap.toArray(Integer.MAX_VALUE)).containsExactly(1, 3, 5); + } + + @Test + public void testDeserializeByteArrayRegion() throws Exception { + RoaringNavigableMap64 expected = RoaringNavigableMap64.bitmapOf(1, 3, 1L << 33); + byte[] serialized = expected.serialize(); + byte[] framed = new byte[serialized.length + 4]; + System.arraycopy(serialized, 0, framed, 2, serialized.length); + + RoaringNavigableMap64 actual = new RoaringNavigableMap64(); + actual.deserialize(framed, 2, serialized.length); + assertThat(actual).isEqualTo(expected); + + assertThatThrownBy(() -> actual.deserialize(framed, 2, framed.length)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Invalid bitmap byte range"); + } + @Test public void testToRangeListAcrossHighBitmapBoundary() { RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
