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 9489f9a3eb [core] Compare composite BTree keys directly from slices
(#10349)
9489f9a3eb is described below
commit 9489f9a3eb094fc4bbec6b93b22dd85a95903d66
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 3 19:17:59 2026 +0800
[core] Compare composite BTree keys directly from slices (#10349)
---
.../paimon/globalindex/CompositeKeySerializer.java | 5 +
.../apache/paimon/globalindex/KeySerializer.java | 6 +
.../paimon/globalindex/btree/BTreeIndexReader.java | 8 +-
.../globalindex/CompositeKeySerializerTest.java | 198 +++++++++++++++++++++
.../globalindex/btree/CompositeBTreeIndexTest.java | 40 +++++
5 files changed, 250 insertions(+), 7 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/CompositeKeySerializer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/CompositeKeySerializer.java
index 8b9ac90276..dba8053f9d 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/CompositeKeySerializer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/CompositeKeySerializer.java
@@ -66,6 +66,11 @@ public class CompositeKeySerializer implements KeySerializer
{
return serializer.get().deserialize(data.copyBytes());
}
+ @Override
+ public Comparator<MemorySlice> createSliceComparator() {
+ return serializer.get().createSliceComparator();
+ }
+
@Override
public Comparator<Object> createComparator() {
return (left, right) -> {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java
index a58b573d9d..5de467df49 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java
@@ -55,6 +55,12 @@ public interface KeySerializer {
Comparator<Object> createComparator();
+ /** Compares serialized keys using the same ordering as {@link
#createComparator()}. */
+ default Comparator<MemorySlice> createSliceComparator() {
+ Comparator<Object> comparator = createComparator();
+ return (left, right) -> comparator.compare(deserialize(left),
deserialize(right));
+ }
+
static KeySerializer create(DataType type) {
return type.accept(
new DataTypeDefaultVisitor<KeySerializer>() {
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 2ab03fb49e..ce0ed043de 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
@@ -181,7 +181,7 @@ public class BTreeIndexReader implements Closeable {
input, filePath, cacheManager,
footer.getBloomFilterHandle());
this.reader =
new SstFileReader(
- createSliceComparator(keySerializer),
+ keySerializer.createSliceComparator(),
blockCache,
footer.getIndexBlockHandle(),
bloomFilter);
@@ -240,12 +240,6 @@ public class BTreeIndexReader implements Closeable {
return nullBitmap;
}
- private Comparator<MemorySlice> createSliceComparator(KeySerializer
keySerializer) {
- return (slice1, slice2) ->
- comparator.compare(
- keySerializer.deserialize(slice1),
keySerializer.deserialize(slice2));
- }
-
@Override
public void close() throws IOException {
// input is this reader's own handle, so it has to be released even
when the reader
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/CompositeKeySerializerTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/CompositeKeySerializerTest.java
new file mode 100644
index 0000000000..32ef5fa65d
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/CompositeKeySerializerTest.java
@@ -0,0 +1,198 @@
+/*
+ * 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.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for serialized global index key ordering. */
+class CompositeKeySerializerTest {
+
+ @ParameterizedTest
+ @MethodSource("typesAndSortedValues")
+ void testCompositeSliceOrdering(DataType type, List<Object> values) {
+ KeySerializer serializer =
+ new CompositeKeySerializer(RowType.of(DataTypes.STRING(),
type, DataTypes.INT()));
+ List<GenericRow> keys = new ArrayList<>();
+ keys.add(GenericRow.of(null, null, null));
+ char[] prefix = new char[160];
+ Arrays.fill(prefix, 'a');
+ BinaryString leading = BinaryString.fromString(new String(prefix));
+ List<Object> nullable = new ArrayList<>();
+ nullable.add(null);
+ nullable.addAll(values);
+ for (Object value : nullable) {
+ keys.add(GenericRow.of(leading, value, null));
+ keys.add(GenericRow.of(leading, value, -1));
+ keys.add(GenericRow.of(leading, value, 1));
+ }
+ keys.add(GenericRow.of(BinaryString.fromString("z"), null, null));
+ assertOrdering(serializer, keys);
+ }
+
+ @ParameterizedTest
+ @MethodSource("typesAndSortedValues")
+ void testScalarSliceOrdering(DataType type, List<Object> values) {
+ assertOrdering(KeySerializer.create(type), values);
+ }
+
+ @Test
+ void testSliceComparisonDoesNotDeserializeRows() {
+ KeySerializer serializer =
+ new CompositeKeySerializer(RowType.of(DataTypes.STRING(),
DataTypes.INT())) {
+ @Override
+ public Object deserialize(MemorySlice data) {
+ throw new AssertionError("Slice comparison must not
deserialize a row.");
+ }
+ };
+ assertOrdering(
+ serializer,
+ Arrays.asList(
+ GenericRow.of(null, null),
+ GenericRow.of(BinaryString.fromString("a"), -1),
+ GenericRow.of(BinaryString.fromString("a"), 1),
+ GenericRow.of(BinaryString.fromString("b"), -1)));
+ }
+
+ private void assertOrdering(KeySerializer serializer, List<?> keys) {
+ Comparator<MemorySlice> comparator =
serializer.createSliceComparator();
+ List<MemorySlice> left = new ArrayList<>();
+ List<MemorySlice> right = new ArrayList<>();
+ for (Object key : keys) {
+ byte[] bytes = serializer.serialize(key);
+ left.add(sliceAtOffset(bytes, 5));
+ right.add(sliceAtOffset(bytes, 13));
+ }
+ for (int i = 0; i < keys.size(); i++) {
+ for (int j = 0; j < keys.size(); j++) {
+ assertThat(Integer.signum(comparator.compare(left.get(i),
right.get(j))))
+ .as("ordering of %s and %s", keys.get(i), keys.get(j))
+ .isEqualTo(Integer.signum(i - j));
+ }
+ }
+ }
+
+ private MemorySlice sliceAtOffset(byte[] bytes, int offset) {
+ byte[] padded = new byte[offset + bytes.length + 7];
+ Arrays.fill(padded, (byte) 0xff);
+ System.arraycopy(bytes, 0, padded, offset, bytes.length);
+ return MemorySlice.wrap(padded).slice(offset, bytes.length);
+ }
+
+ private static Stream<Arguments> typesAndSortedValues() {
+ List<Object> strings =
+ Arrays.asList(
+ BinaryString.fromString(""),
+ BinaryString.fromString("a"),
+ BinaryString.fromString("a\u0000b"),
+ BinaryString.fromString("aa"),
+ BinaryString.fromString("b"),
+ BinaryString.fromString("中文"));
+ List<Object> timestamps =
+ Arrays.asList(
+ Timestamp.fromEpochMillis(-1),
+ Timestamp.fromEpochMillis(0),
+ Timestamp.fromEpochMillis(1));
+ List<Object> preciseTimestamps =
+ Arrays.asList(
+ Timestamp.fromEpochMillis(-1, 999999),
+ Timestamp.fromEpochMillis(0),
+ Timestamp.fromEpochMillis(0, 1),
+ Timestamp.fromEpochMillis(0, 999999),
+ Timestamp.fromEpochMillis(1));
+ return Stream.of(
+ Arguments.of(DataTypes.BOOLEAN(), Arrays.asList(false, true)),
+ Arguments.of(
+ DataTypes.TINYINT(),
+ Arrays.asList(Byte.MIN_VALUE, (byte) 0,
Byte.MAX_VALUE)),
+ Arguments.of(
+ DataTypes.SMALLINT(),
+ Arrays.asList(Short.MIN_VALUE, (short) 0,
Short.MAX_VALUE)),
+ Arguments.of(
+ DataTypes.INT(),
+ Arrays.asList(Integer.MIN_VALUE, -1, 0, 256,
Integer.MAX_VALUE)),
+ Arguments.of(
+ DataTypes.BIGINT(), Arrays.asList(Long.MIN_VALUE, -1L,
0L, Long.MAX_VALUE)),
+ Arguments.of(
+ DataTypes.FLOAT(),
+ Arrays.asList(
+ Float.NEGATIVE_INFINITY,
+ -1.0f,
+ -0.0f,
+ 0.0f,
+ Float.POSITIVE_INFINITY,
+ Float.NaN)),
+ Arguments.of(
+ DataTypes.DOUBLE(),
+ Arrays.asList(
+ Double.NEGATIVE_INFINITY,
+ -1.0d,
+ -0.0d,
+ 0.0d,
+ Double.POSITIVE_INFINITY,
+ Double.NaN)),
+ Arguments.of(DataTypes.CHAR(256), strings),
+ Arguments.of(DataTypes.STRING(), strings),
+ Arguments.of(
+ DataTypes.DECIMAL(18, 3),
+ decimals(18, 3, "-999999999999999.999", "0",
"999999999999999.999")),
+ Arguments.of(
+ DataTypes.DECIMAL(38, 18),
+ decimals(
+ 38,
+ 18,
+ "-99999999999999999999.999999999999999999",
+ "0",
+ "99999999999999999999.999999999999999999")),
+ Arguments.of(DataTypes.DATE(), Arrays.asList(-1, 0, 1)),
+ Arguments.of(DataTypes.TIME(), Arrays.asList(0, 1, 86399999)),
+ Arguments.of(DataTypes.TIMESTAMP(3), timestamps),
+ Arguments.of(DataTypes.TIMESTAMP(9), preciseTimestamps),
+ Arguments.of(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(3),
timestamps),
+ Arguments.of(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(9),
preciseTimestamps));
+ }
+
+ private static List<Object> decimals(int precision, int scale, String...
values) {
+ List<Object> result = new ArrayList<>();
+ for (String value : values) {
+ result.add(Decimal.fromBigDecimal(new BigDecimal(value),
precision, scale));
+ }
+ return result;
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
index df3975a4f2..6733b3af88 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
@@ -34,7 +34,9 @@ import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.SortedGlobalIndexer;
import org.apache.paimon.globalindex.SortedIndexFileMeta;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -60,6 +62,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
import static org.assertj.core.api.Assertions.assertThat;
@@ -83,6 +86,7 @@ class CompositeBTreeIndexTest {
options.set(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION, version);
options.set(BTreeIndexOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED, true);
options.set(BTreeIndexOptions.BTREE_INDEX_COMPRESSION, "lz4");
+ options.set(BTreeIndexOptions.BTREE_INDEX_BLOCK_SIZE,
MemorySize.ofBytes(64));
GlobalIndexer indexer = GlobalIndexer.create("btree",
type.getFields(), options);
LocalFileIO io = LocalFileIO.create();
Path directory = new Path(tempPath.toUri());
@@ -125,6 +129,33 @@ class CompositeBTreeIndexTest {
Path path = new Path(directory, result.fileName());
GlobalIndexIOMeta meta =
new GlobalIndexIOMeta(path, io.getFileSize(path),
result.rowCount(), result.meta());
+ AtomicInteger deserializations = new AtomicInteger();
+ KeySerializer countingSerializer =
+ new CompositeKeySerializer(type) {
+ @Override
+ public Object deserialize(MemorySlice data) {
+ deserializations.incrementAndGet();
+ return super.deserialize(data);
+ }
+ };
+ try (CacheManager cache = new CacheManager(MemorySize.ofMebiBytes(1),
0.1);
+ BTreeIndexReader reader =
+ new BTreeIndexReader(
+ countingSerializer,
+ file -> io.newInputStream(file.filePath()),
+ meta,
+ cache,
+ null)) {
+ int onOpen = deserializations.get();
+ assertThat(reader.visitEqual(row("category-a", -1,
"")).get().results().toRangeList())
+ .containsExactly(new Range(0, 0));
+ assertThat(reader.visitEqual(row("category-a", 7,
"tag")).get().results().toRangeList())
+ .containsExactly(new Range(2, 3));
+ assertThat(reader.visitEqual(row("category-b", 7,
"tag")).get().results().toRangeList())
+ .containsExactly(new Range(4, 4));
+ assertThat(reader.visitEqual(row("category-a", 8,
"tag")).get().results()).isEmpty();
+ assertThat(deserializations.get()).isEqualTo(onOpen);
+ }
assertThat(
new BTreeGlobalIndexerFactory()
.selectFiles(
@@ -364,6 +395,7 @@ class CompositeBTreeIndexTest {
void testConcurrentRoundTrips() throws Exception {
RowType type = RowType.of(DataTypes.STRING(), DataTypes.INT(),
DataTypes.STRING());
KeySerializer serializer = new CompositeKeySerializer(type);
+ Comparator<MemorySlice> sliceComparator =
serializer.createSliceComparator();
ExecutorService executor = Executors.newFixedThreadPool(8);
CountDownLatch start = new CountDownLatch(1);
List<Future<?>> tasks = new ArrayList<>();
@@ -386,6 +418,14 @@ class CompositeBTreeIndexTest {
Object restored =
serializer.deserialize(MemorySlice.wrap(bytes));
assertThat(restored).isEqualTo(key);
+ byte[] next =
+ serializer.serialize(
+ row("category-" +
worker, i + 1, null));
+ assertThat(
+
sliceComparator.compare(
+
MemorySlice.wrap(bytes),
+
MemorySlice.wrap(next)))
+ .isNegative();
assertThat(
serializer
.createComparator()
