[kv] Use JDK9 Arrays#equals() to check bytes prefix for PrefixLookup (#…

…331)

fluss-common/src/main/java/com/alibaba/fluss/utils/BytesUtils.java

@@ -17,8 +17,11 @@
 package com.alibaba.fluss.utils;
 
 import com.alibaba.fluss.annotation.Internal;
+import com.alibaba.fluss.utils.crc.Java;
 
+import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 
 /** Utils for bytes. */
 @Internal
@@ -58,4 +61,100 @@ public static byte[] toArray(ByteBuffer buffer, int offset, int size) {
         }
         return dest;
     }
+
+    /**
+     * Check if the given first byte array ({@code prefix}) is a prefix of the second byte array
+     * ({@code bytes}).
+     *
+     * @param prefix the prefix byte array
+     * @param bytes the byte array to check if it has the prefix
+     * @return true if the given bytes has the given prefix, false otherwise.
+     */
+    public static boolean prefixEquals(byte[] prefix, byte[] bytes) {
+        return BEST_EQUAL_COMPARER.prefixEquals(prefix, bytes);
+    }
+
+    // -------------------------------------------------------------------------------------------
+
+    private static final BytesPrefixComparer BEST_EQUAL_COMPARER;
+
+    static {
+        if (Java.IS_JAVA9_COMPATIBLE) {
+            BEST_EQUAL_COMPARER = new Java9BytesPrefixComparer();
+        } else {
+            BEST_EQUAL_COMPARER = new PureJavaBytesPrefixComparer();
+        }
+    }
+
+    /** Compare two byte arrays for equality. */
+    private interface BytesPrefixComparer {
+
+        /**
+         * Check if the given first byte array ({@code prefix}) is a prefix of the second byte array
+         * ({@code bytes}).
+         *
+         * @param prefix The prefix byte array
+         * @param bytes The byte array to check if it has the prefix
+         * @return true if the given bytes has the given prefix, false otherwise
+         */
+        boolean prefixEquals(byte[] prefix, byte[] bytes);
+    }
+
+    private static final class Java9BytesPrefixComparer implements BytesPrefixComparer {
+        private static final Method EQUALS_METHOD;
+
+        static {
+            try {
+                EQUALS_METHOD =
+                        Class.forName(Arrays.class.getName())
+                                .getMethod(
+                                        "equals",
+                                        byte[].class,
+                                        int.class,
+                                        int.class,
+                                        byte[].class,
+                                        int.class,
+                                        int.class);
+            } catch (Exception e) {
+                throw new RuntimeException("Failed to load Arrays.equals method", e);
+            }
+        }
+
+        @Override
+        public boolean prefixEquals(byte[] prefix, byte[] bytes) {
+            if (prefix.length > bytes.length) {
+                return false;
+            }
+            try {
+                int fromIndex = 0; // inclusive
+                int toIndex = prefix.length; // exclusive
+                return (boolean)
+                        EQUALS_METHOD.invoke(
+                                null, prefix, fromIndex, toIndex, bytes, fromIndex, toIndex);
+            } catch (Throwable e) {
+                // should never happen
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+    /**
+     * A pure Java implementation of the {@link BytesPrefixComparer} that does not rely on Java 9
+     * APIs and compares bytes one by one which is slower.
+     */
+    private static final class PureJavaBytesPrefixComparer implements BytesPrefixComparer {
+
+        @Override
+        public boolean prefixEquals(byte[] prefix, byte[] bytes) {
+            if (prefix.length > bytes.length) {
+                return false;
+            }
+            for (int i = 0; i < prefix.length; i++) {
+                if (prefix[i] != bytes[i]) {
+                    return false;
+                }
+            }
+            return true;
+        }
+    }
 }

fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBKv.java

@@ -19,6 +19,7 @@
 import com.alibaba.fluss.exception.FlussRuntimeException;
 import com.alibaba.fluss.rocksdb.RocksDBOperationUtils;
 import com.alibaba.fluss.server.utils.ResourceGuard;
+import com.alibaba.fluss.utils.BytesUtils;
 import com.alibaba.fluss.utils.IOUtils;
 
 import org.rocksdb.ColumnFamilyHandle;
@@ -106,7 +107,7 @@ public List<byte[]> prefixLookup(byte[] prefixKey) {
         RocksIterator iterator = db.newIterator(defaultColumnFamilyHandle, readOptions);
         try {
             iterator.seek(prefixKey);
-            while (iterator.isValid() && isPrefixEquals(prefixKey, iterator.key())) {
+            while (iterator.isValid() && BytesUtils.prefixEquals(prefixKey, iterator.key())) {
                 pkList.add(iterator.value());
                 iterator.next();
             }
@@ -204,28 +205,4 @@ public void close() throws Exception {
     public RocksDB getDb() {
         return db;
     }
-
-    /**
-     * Check if the given first byte array ({@code prefix}) is a prefix of the second byte array
-     * ({@code bytes}).
-     *
-     * @param prefix The prefix byte array
-     * @param bytes The byte array to check if it has the prefix
-     * @return true if the given bytes has the given prefix, false otherwise
-     */
-    public static boolean isPrefixEquals(byte[] prefix, byte[] bytes) {
-        // TODO, This is very inefficient to compare arrays byte by byte. In the future we can
-        // use JDK9 Arrays.compare(compare(byte[] a, int aFromIndex, int aToIndex, byte[] b, int
-        // bFromIndex, int bToIndex)) to instead. See issue:
-        // https://github.com/alibaba/fluss/issues/271
-        if (prefix.length > bytes.length) {
-            return false;
-        }
-        for (int i = 0; i < prefix.length; i++) {
-            if (prefix[i] != bytes[i]) {
-                return false;
-            }
-        }
-        return true;
-    }
 }