This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a905466d7b [core] Add adaptive BTree posting list encoding (#9981)
a905466d7b is described below
commit a905466d7bf49d80162b6248fc80a2f35fae7d44
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();
