[common] Accelerate endsWith and like '%x' with reverse btree global index

paimon-common/src/main/java/org/apache/paimon/globalindex/ReversedKeySerializer.java

@@ -0,0 +1,67 @@
+/*
+ * 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.memory.MemorySlice;
+
+import java.util.Comparator;
+
+/** A {@link KeySerializer} that reverses serialized key bytes. */
+public class ReversedKeySerializer implements KeySerializer {
+
+    private final KeySerializer delegate;
+
+    public ReversedKeySerializer(KeySerializer delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public byte[] serialize(Object key) {
+        return reverse(delegate.serialize(key));
+    }
+
+    @Override
+    public Object deserialize(MemorySlice data) {
+        return delegate.deserialize(MemorySlice.wrap(reverse(data.copyBytes())));
+    }
+
+    @Override
+    public Comparator<Object> createComparator() {
+        return (left, right) -> compareUnsigned(serialize(left), serialize(right));
+    }
+
+    private static byte[] reverse(byte[] bytes) {
+        byte[] out = new byte[bytes.length];
+        for (int i = 0; i < bytes.length; i++) {
+            out[i] = bytes[bytes.length - 1 - i];
+        }
+        return out;
+    }
+
+    private static int compareUnsigned(byte[] left, byte[] right) {
+        int len = Math.min(left.length, right.length);
+        for (int i = 0; i < len; i++) {
+            int cmp = (left[i] & 0xFF) - (right[i] & 0xFF);
+            if (cmp != 0) {
+                return cmp;
+            }
+        }
+        return left.length - right.length;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java

@@ -66,7 +66,11 @@ public class BTreeGlobalIndexer implements GlobalIndexer {
     private final LazyField<CacheManager> cacheManager;
 
     public BTreeGlobalIndexer(DataField dataField, Options options) {
-        this.keySerializer = KeySerializer.create(dataField.type());
+        this(KeySerializer.create(dataField.type()), options);
+    }
+
+    public BTreeGlobalIndexer(KeySerializer keySerializer, Options options) {
+        this.keySerializer = keySerializer;
         this.options = options;
         this.fallbackScanMaxSize =
                 options.get(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE).getBytes();
@@ -99,12 +103,23 @@ public GlobalIndexReader createReader(
             GlobalIndexFileReader fileReader,
             List<GlobalIndexIOMeta> files,
             ExecutorService executor) {
-        return new LazyFilteredBTreeReader(
+        return newReader(
                 files,
                 keySerializer,
                 fileReader,
                 cacheManager.get(),
                 fallbackScanMaxSize,
                 executor);
     }
+
+    protected LazyFilteredBTreeReader newReader(
+            List<GlobalIndexIOMeta> files,
+            KeySerializer keySerializer,
+            GlobalIndexFileReader fileReader,
+            CacheManager cacheManager,
+            long fallbackScanMaxSize,
+            ExecutorService executor) {
+        return new LazyFilteredBTreeReader(
+                files, keySerializer, fileReader, cacheManager, fallbackScanMaxSize, executor);
+    }
 }

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/ReverseBTreeGlobalIndexer.java

@@ -0,0 +1,50 @@
+/*
+ * 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.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.ReversedKeySerializer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.io.cache.CacheManager;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+/** A btree global index over reversed keys. */
+public class ReverseBTreeGlobalIndexer extends BTreeGlobalIndexer {
+
+    public ReverseBTreeGlobalIndexer(DataField dataField, Options options) {
+        super(new ReversedKeySerializer(KeySerializer.create(dataField.type())), options);
+    }
+
+    @Override
+    protected LazyFilteredBTreeReader newReader(
+            List<GlobalIndexIOMeta> files,
+            KeySerializer keySerializer,
+            GlobalIndexFileReader fileReader,
+            CacheManager cacheManager,
+            long fallbackScanMaxSize,
+            ExecutorService executor) {
+        return new ReverseLazyFilteredBTreeReader(
+                files, keySerializer, fileReader, cacheManager, fallbackScanMaxSize, executor);
+    }
+}

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/ReverseBTreeGlobalIndexerFactory.java

@@ -0,0 +1,40 @@
+/*
+ * 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.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.GlobalIndexerFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+
+/** Factory for {@link ReverseBTreeGlobalIndexer}. */
+public class ReverseBTreeGlobalIndexerFactory implements GlobalIndexerFactory {
+
+    public static final String IDENTIFIER = "reverse-btree";
+
+    @Override
+    public String identifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public GlobalIndexer create(DataField dataField, Options options) {
+        return new ReverseBTreeGlobalIndexer(dataField, options);
+    }
+}

paimon-common/src/main/java/org/apache/paimon/globalindex/btree/ReverseLazyFilteredBTreeReader.java

@@ -0,0 +1,97 @@
+/*
+ * 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.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.io.cache.CacheManager;
+import org.apache.paimon.predicate.FieldRef;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+/** A {@link LazyFilteredBTreeReader} over reversed keys. */
+public class ReverseLazyFilteredBTreeReader extends LazyFilteredBTreeReader {
+
+    public ReverseLazyFilteredBTreeReader(
+            List<GlobalIndexIOMeta> files,
+            KeySerializer keySerializer,
+            GlobalIndexFileReader fileReader,
+            CacheManager cacheManager,
+            long fallbackScanMaxSize,
+            ExecutorService executor) {
+        super(files, keySerializer, fileReader, cacheManager, fallbackScanMaxSize, executor);
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+            FieldRef fieldRef, Object literal) {
+        return super.visitStartsWith(fieldRef, literal);
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+            FieldRef fieldRef, Object literal) {
+        return unsupported();
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+            FieldRef fieldRef, Object literal) {
+        return unsupported();
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+            FieldRef fieldRef, Object literal) {
+        return unsupported();
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+            FieldRef fieldRef, Object literal) {
+        return unsupported();
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual(
+            FieldRef fieldRef, Object literal) {
+        return unsupported();
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+            FieldRef fieldRef, Object from, Object to) {
+        return unsupported();
+    }
+
+    @Override
+    public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
+            FieldRef fieldRef, Object from, Object to) {
+        return unsupported();
+    }
+
+    private static CompletableFuture<Optional<GlobalIndexResult>> unsupported() {
+        return CompletableFuture.completedFuture(Optional.empty());
+    }
+}

paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory

@@ -14,4 +14,5 @@
 # limitations under the License.
 
 org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory
+org.apache.paimon.globalindex.btree.ReverseBTreeGlobalIndexerFactory
 org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory

paimon-common/src/test/java/org/apache/paimon/globalindex/ReversedKeySerializerTest.java

@@ -0,0 +1,63 @@
+/*
+ * 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.memory.MemorySlice;
+import org.apache.paimon.types.VarCharType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ReversedKeySerializer}. */
+public class ReversedKeySerializerTest {
+
+    private final KeySerializer serializer =
+            new ReversedKeySerializer(
+                    KeySerializer.create(new VarCharType(VarCharType.MAX_LENGTH)));
+
+    @Test
+    public void testRoundTrip() {
+        for (String s : new String[] {"", "a", "hello", "ends_with_suffix"}) {
+            BinaryString key = BinaryString.fromString(s);
+            assertThat(serializer.deserialize(MemorySlice.wrap(serializer.serialize(key))))
+                    .isEqualTo(key);
+        }
+    }
+
+    @Test
+    public void testComparatorOrdersBySuffix() {
+        Comparator<Object> cmp = serializer.createComparator();
+        List<BinaryString> input =
+                Arrays.asList(
+                        BinaryString.fromString("2b"),
+                        BinaryString.fromString("1a"),
+                        BinaryString.fromString("2a"),
+                        BinaryString.fromString("1b"));
+        List<String> sorted =
+                input.stream().sorted(cmp).map(BinaryString::toString).collect(Collectors.toList());
+        assertThat(sorted).containsExactly("1a", "2a", "1b", "2b");
+    }
+}