Improve RocksJava Comparator (#6252)

Summary:
This is a redesign of the API for RocksJava comparators with the aim of improving performance. It also simplifies the class hierarchy.

**NOTE**: This breaks backwards compatibility for existing 3rd party Comparators implemented in Java... so we need to consider carefully which release branches this goes into.

Previously when implementing a comparator in Java the developer had a choice of subclassing either `DirectComparator` or `Comparator` which would use direct and non-direct byte-buffers resepectively (via `DirectSlice` and `Slice`).

In this redesign there we have eliminated the overhead of using the Java Slice classes, and just use `ByteBuffer`s. The `ComparatorOptions` supplied when constructing a Comparator allow you to choose between direct and non-direct byte buffers by setting `useDirect`.

In addition, the `ComparatorOptions` now allow you to choose whether a ByteBuffer is reused over multiple comparator calls, by setting `maxReusedBufferSize > 0`. When buffers are reused, ComparatorOptions provides a choice of mutex type by setting `useAdaptiveMutex`.

 ---
[JMH benchmarks previously indicated](#6241 (comment)) that the difference between C++ and Java for implementing a comparator was ~7x slowdown in Java.

With these changes, when reusing buffers and guarding access to them via mutexes the slowdown is approximately the same. However, these changes offer a new facility to not reuse mutextes, which reduces the slowdown to ~5.5x in Java. We also offer a `thread_local` mechanism for reusing buffers, which reduces slowdown to ~5.2x in Java (closes #4425).

These changes also form a good base for further optimisation work such as further JNI lookup caching, and JNI critical.

 ---
These numbers were captured without jemalloc. With jemalloc, the performance improves for all tests, and the Java slowdown reduces to between 4.8x and 5.x.

```
ComparatorBenchmarks.put                                                native_bytewise  thrpt   25  124483.795 ± 2032.443  ops/s
ComparatorBenchmarks.put                                        native_reverse_bytewise  thrpt   25  114414.536 ± 3486.156  ops/s
ComparatorBenchmarks.put              java_bytewise_non-direct_reused-64_adaptive-mutex  thrpt   25   17228.250 ± 1288.546  ops/s
ComparatorBenchmarks.put          java_bytewise_non-direct_reused-64_non-adaptive-mutex  thrpt   25   16035.865 ± 1248.099  ops/s
ComparatorBenchmarks.put                java_bytewise_non-direct_reused-64_thread-local  thrpt   25   21571.500 ±  871.521  ops/s
ComparatorBenchmarks.put                  java_bytewise_direct_reused-64_adaptive-mutex  thrpt   25   23613.773 ± 8465.660  ops/s
ComparatorBenchmarks.put              java_bytewise_direct_reused-64_non-adaptive-mutex  thrpt   25   16768.172 ± 5618.489  ops/s
ComparatorBenchmarks.put                    java_bytewise_direct_reused-64_thread-local  thrpt   25   23921.164 ± 8734.742  ops/s
ComparatorBenchmarks.put                              java_bytewise_non-direct_no-reuse  thrpt   25   17899.684 ±  839.679  ops/s
ComparatorBenchmarks.put                                  java_bytewise_direct_no-reuse  thrpt   25   22148.316 ± 1215.527  ops/s
ComparatorBenchmarks.put      java_reverse_bytewise_non-direct_reused-64_adaptive-mutex  thrpt   25   11311.126 ±  820.602  ops/s
ComparatorBenchmarks.put  java_reverse_bytewise_non-direct_reused-64_non-adaptive-mutex  thrpt   25   11421.311 ±  807.210  ops/s
ComparatorBenchmarks.put        java_reverse_bytewise_non-direct_reused-64_thread-local  thrpt   25   11554.005 ±  960.556  ops/s
ComparatorBenchmarks.put          java_reverse_bytewise_direct_reused-64_adaptive-mutex  thrpt   25   22960.523 ± 1673.421  ops/s
ComparatorBenchmarks.put      java_reverse_bytewise_direct_reused-64_non-adaptive-mutex  thrpt   25   18293.317 ± 1434.601  ops/s
ComparatorBenchmarks.put            java_reverse_bytewise_direct_reused-64_thread-local  thrpt   25   24479.361 ± 2157.306  ops/s
ComparatorBenchmarks.put                      java_reverse_bytewise_non-direct_no-reuse  thrpt   25    7942.286 ±  626.170  ops/s
ComparatorBenchmarks.put                          java_reverse_bytewise_direct_no-reuse  thrpt   25   11781.955 ± 1019.843  ops/s
```
Pull Request resolved: #6252

Differential Revision: D19331064

Pulled By: pdillinger

fbshipit-source-id: 1f3b794e6a14162b2c3ffb943e8c0e64a0c03738

HISTORY.md

@@ -1,5 +1,8 @@
 # Rocksdb Change Log
 ## Unreleased
+### Public API Change
+* Major breaking changes to Java comparators, toward standardizing on ByteBuffer for performant, locale-neutral operations on keys (#6252).
+
 ### Bug Fixes
 * Fix incorrect results while block-based table uses kHashSearch, together with Prev()/SeekForPrev().
 * Fix a bug that prevents opening a DB after two consecutive crash with TransactionDB, where the first crash recovers from a corrupted WAL with kPointInTimeRecovery but the second cannot.

java/CMakeLists.txt

@@ -122,7 +122,6 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/CompactRangeOptions.java
   src/main/java/org/rocksdb/CompactionStopStyle.java
   src/main/java/org/rocksdb/CompactionStyle.java
-  src/main/java/org/rocksdb/Comparator.java
   src/main/java/org/rocksdb/ComparatorOptions.java
   src/main/java/org/rocksdb/ComparatorType.java
   src/main/java/org/rocksdb/CompressionOptions.java
@@ -131,7 +130,6 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/DBOptionsInterface.java
   src/main/java/org/rocksdb/DBOptions.java
   src/main/java/org/rocksdb/DbPath.java
-  src/main/java/org/rocksdb/DirectComparator.java
   src/main/java/org/rocksdb/DirectSlice.java
   src/main/java/org/rocksdb/EncodingType.java
   src/main/java/org/rocksdb/Env.java
@@ -181,6 +179,7 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/ReadTier.java
   src/main/java/org/rocksdb/RemoveEmptyValueCompactionFilter.java
   src/main/java/org/rocksdb/RestoreOptions.java
+  src/main/java/org/rocksdb/ReusedSynchronisationType.java
   src/main/java/org/rocksdb/RocksCallbackObject.java
   src/main/java/org/rocksdb/RocksDBException.java
   src/main/java/org/rocksdb/RocksDB.java
@@ -236,9 +235,10 @@ set(JAVA_MAIN_CLASSES
   src/main/java/org/rocksdb/WriteBatchWithIndex.java
   src/main/java/org/rocksdb/WriteOptions.java
   src/main/java/org/rocksdb/WriteBufferManager.java
+  src/main/java/org/rocksdb/util/ByteUtil.java
   src/main/java/org/rocksdb/util/BytewiseComparator.java
-  src/main/java/org/rocksdb/util/DirectBytewiseComparator.java
   src/main/java/org/rocksdb/util/Environment.java
+  src/main/java/org/rocksdb/util/IntComparator.java
   src/main/java/org/rocksdb/util/ReverseBytewiseComparator.java
   src/main/java/org/rocksdb/util/SizeUnit.java
   src/main/java/org/rocksdb/UInt64AddOperator.java
@@ -400,11 +400,9 @@ if(${CMAKE_VERSION} VERSION_LESS "3.11.4" OR (${Java_VERSION_MINOR} STREQUAL "7"
           org.rocksdb.CompactionOptionsFIFO
           org.rocksdb.CompactionOptionsUniversal
           org.rocksdb.CompactRangeOptions
-          org.rocksdb.Comparator
           org.rocksdb.ComparatorOptions
           org.rocksdb.CompressionOptions
           org.rocksdb.DBOptions
-          org.rocksdb.DirectComparator
           org.rocksdb.DirectSlice
           org.rocksdb.Env
           org.rocksdb.EnvOptions

java/Makefile

@@ -1,5 +1,7 @@
-NATIVE_JAVA_CLASSES = org.rocksdb.AbstractCompactionFilter\
+NATIVE_JAVA_CLASSES = \
+	org.rocksdb.AbstractCompactionFilter\
 	org.rocksdb.AbstractCompactionFilterFactory\
+	org.rocksdb.AbstractComparator\
 	org.rocksdb.AbstractSlice\
 	org.rocksdb.AbstractTableFilter\
 	org.rocksdb.AbstractTraceWriter\
@@ -21,11 +23,9 @@ NATIVE_JAVA_CLASSES = org.rocksdb.AbstractCompactionFilter\
 	org.rocksdb.CompactionOptionsFIFO\
 	org.rocksdb.CompactionOptionsUniversal\
 	org.rocksdb.CompactRangeOptions\
-	org.rocksdb.Comparator\
 	org.rocksdb.ComparatorOptions\
 	org.rocksdb.CompressionOptions\
 	org.rocksdb.DBOptions\
-	org.rocksdb.DirectComparator\
 	org.rocksdb.DirectSlice\
 	org.rocksdb.Env\
 	org.rocksdb.EnvOptions\
@@ -98,10 +98,13 @@ ifeq ($(PLATFORM), OS_MACOSX)
 ROCKSDB_JAR = rocksdbjni-$(ROCKSDB_MAJOR).$(ROCKSDB_MINOR).$(ROCKSDB_PATCH)-osx.jar
 endif
 
-JAVA_TESTS = org.rocksdb.BackupableDBOptionsTest\
+JAVA_TESTS = \
+	org.rocksdb.BackupableDBOptionsTest\
 	org.rocksdb.BackupEngineTest\
 	org.rocksdb.BlockBasedTableConfigTest\
+	org.rocksdb.BuiltinComparatorTest\
 	org.rocksdb.util.BytewiseComparatorTest\
+	org.rocksdb.util.BytewiseComparatorIntTest\
 	org.rocksdb.CheckPointTest\
 	org.rocksdb.ClockCacheTest\
 	org.rocksdb.ColumnFamilyOptionsTest\
@@ -115,16 +118,16 @@ JAVA_TESTS = org.rocksdb.BackupableDBOptionsTest\
 	org.rocksdb.CompactionPriorityTest\
 	org.rocksdb.CompactionStopStyleTest\
 	org.rocksdb.ComparatorOptionsTest\
-	org.rocksdb.ComparatorTest\
 	org.rocksdb.CompressionOptionsTest\
 	org.rocksdb.CompressionTypesTest\
 	org.rocksdb.DBOptionsTest\
-	org.rocksdb.DirectComparatorTest\
 	org.rocksdb.DirectSliceTest\
+	org.rocksdb.util.EnvironmentTest\
 	org.rocksdb.EnvOptionsTest\
 	org.rocksdb.HdfsEnvTest\
 	org.rocksdb.IngestExternalFileOptionsTest\
-	org.rocksdb.util.EnvironmentTest\
+	org.rocksdb.util.IntComparatorTest\
+	org.rocksdb.util.JNIComparatorTest\
 	org.rocksdb.FilterTest\
 	org.rocksdb.FlushTest\
 	org.rocksdb.InfoLogLevelTest\
@@ -148,6 +151,7 @@ JAVA_TESTS = org.rocksdb.BackupableDBOptionsTest\
 	org.rocksdb.RateLimiterTest\
 	org.rocksdb.ReadOnlyTest\
 	org.rocksdb.ReadOptionsTest\
+	org.rocksdb.util.ReverseBytewiseComparatorIntTest\
 	org.rocksdb.RocksDBTest\
 	org.rocksdb.RocksDBExceptionTest\
 	org.rocksdb.DefaultEnvTest\

java/jmh/pom.xml

@@ -50,7 +50,7 @@
         <dependency>
             <groupId>org.rocksdb</groupId>
             <artifactId>rocksdbjni</artifactId>
-            <version>6.4.6</version>
+            <version>6.6.0-SNAPSHOT</version>
         </dependency>
 
         <dependency>

java/jmh/src/main/java/org/rocksdb/jmh/ComparatorBenchmarks.java

@@ -9,7 +9,6 @@
 import org.openjdk.jmh.annotations.*;
 import org.rocksdb.*;
 import org.rocksdb.util.BytewiseComparator;
-import org.rocksdb.util.DirectBytewiseComparator;
 import org.rocksdb.util.FileUtils;
 import org.rocksdb.util.ReverseBytewiseComparator;
 
@@ -26,13 +25,24 @@ public class ComparatorBenchmarks {
   @Param({
       "native_bytewise",
       "native_reverse_bytewise",
-      "java_bytewise_adaptive_mutex",
-      "java_bytewise_non-adaptive_mutex",
-      "java_reverse_bytewise_adaptive_mutex",
-      "java_reverse_bytewise_non-adaptive_mutex",
-      "java_direct_bytewise_adaptive_mutex",
-      "java_direct_bytewise_non-adaptive_mutex"
 
+      "java_bytewise_non-direct_reused-64_adaptive-mutex",
+      "java_bytewise_non-direct_reused-64_non-adaptive-mutex",
+      "java_bytewise_non-direct_reused-64_thread-local",
+      "java_bytewise_direct_reused-64_adaptive-mutex",
+      "java_bytewise_direct_reused-64_non-adaptive-mutex",
+      "java_bytewise_direct_reused-64_thread-local",
+      "java_bytewise_non-direct_no-reuse",
+      "java_bytewise_direct_no-reuse",
+
+      "java_reverse_bytewise_non-direct_reused-64_adaptive-mutex",
+      "java_reverse_bytewise_non-direct_reused-64_non-adaptive-mutex",
+      "java_reverse_bytewise_non-direct_reused-64_thread-local",
+      "java_reverse_bytewise_direct_reused-64_adaptive-mutex",
+      "java_reverse_bytewise_direct_reused-64_non-adaptive-mutex",
+      "java_reverse_bytewise_direct_reused-64_thread-local",
+      "java_reverse_bytewise_non-direct_no-reuse",
+      "java_reverse_bytewise_direct_no-reuse"
   })
   public String comparatorName;
 
@@ -50,42 +60,49 @@ public void setup() throws IOException, RocksDBException {
 
     options = new Options()
             .setCreateIfMissing(true);
-    if (comparatorName == null || "native_bytewise".equals(comparatorName)) {
+
+    if ("native_bytewise".equals(comparatorName)) {
       options.setComparator(BuiltinComparator.BYTEWISE_COMPARATOR);
+
     } else if ("native_reverse_bytewise".equals(comparatorName)) {
       options.setComparator(BuiltinComparator.REVERSE_BYTEWISE_COMPARATOR);
-    } else if ("java_bytewise_adaptive_mutex".equals(comparatorName)) {
-      comparatorOptions = new ComparatorOptions()
-              .setUseAdaptiveMutex(true);
-      comparator = new BytewiseComparator(comparatorOptions);
-      options.setComparator(comparator);
-    } else if ("java_bytewise_non-adaptive_mutex".equals(comparatorName)) {
-      comparatorOptions = new ComparatorOptions()
-              .setUseAdaptiveMutex(false);
-      comparator = new BytewiseComparator(comparatorOptions);
-      options.setComparator(comparator);
-    } else if ("java_reverse_bytewise_adaptive_mutex".equals(comparatorName)) {
-      comparatorOptions = new ComparatorOptions()
-              .setUseAdaptiveMutex(true);
-      comparator = new ReverseBytewiseComparator(comparatorOptions);
-      options.setComparator(comparator);
-    } else if ("java_reverse_bytewise_non-adaptive_mutex".equals(comparatorName)) {
-      comparatorOptions = new ComparatorOptions()
-              .setUseAdaptiveMutex(false);
-      comparator = new ReverseBytewiseComparator(comparatorOptions);
-      options.setComparator(comparator);
-    } else if ("java_direct_bytewise_adaptive_mutex".equals(comparatorName)) {
-      comparatorOptions = new ComparatorOptions()
-              .setUseAdaptiveMutex(true);
-      comparator = new DirectBytewiseComparator(comparatorOptions);
-      options.setComparator(comparator);
-    } else if ("java_direct_bytewise_non-adaptive_mutex".equals(comparatorName)) {
-      comparatorOptions = new ComparatorOptions()
-              .setUseAdaptiveMutex(false);
-      comparator = new DirectBytewiseComparator(comparatorOptions);
+
+    } else if (comparatorName.startsWith("java_")) {
+      comparatorOptions = new ComparatorOptions();
+
+      if (comparatorName.indexOf("non-direct") > -1) {
+        comparatorOptions.setUseDirectBuffer(false);
+      } else if (comparatorName.indexOf("direct") > -1) {
+        comparatorOptions.setUseDirectBuffer(true);
+      }
+
+      if (comparatorName.indexOf("no-reuse") > -1) {
+        comparatorOptions.setMaxReusedBufferSize(-1);
+      } else if (comparatorName.indexOf("_reused-") > -1) {
+        final int idx = comparatorName.indexOf("_reused-");
+        String s = comparatorName.substring(idx + 8);
+        s = s.substring(0, s.indexOf('_'));
+        comparatorOptions.setMaxReusedBufferSize(Integer.parseInt(s));
+      }
+
+      if (comparatorName.indexOf("non-adaptive-mutex") > -1) {
+        comparatorOptions.setReusedSynchronisationType(ReusedSynchronisationType.MUTEX);
+      } else if (comparatorName.indexOf("adaptive-mutex") > -1) {
+        comparatorOptions.setReusedSynchronisationType(ReusedSynchronisationType.ADAPTIVE_MUTEX);
+      } else if (comparatorName.indexOf("thread-local") > -1) {
+        comparatorOptions.setReusedSynchronisationType(ReusedSynchronisationType.THREAD_LOCAL);
+      }
+
+      if (comparatorName.startsWith("java_bytewise")) {
+        comparator = new BytewiseComparator(comparatorOptions);
+      } else if (comparatorName.startsWith("java_reverse_bytewise")) {
+        comparator = new ReverseBytewiseComparator(comparatorOptions);
+      }
+
       options.setComparator(comparator);
+
     } else {
-      throw new IllegalArgumentException("Unknown comparator name: " + comparatorName);
+      throw new IllegalArgumentException("Unknown comparatorName: " + comparatorName);
     }
 
     db = RocksDB.open(options, dbDir.toAbsolutePath().toString());

java/rocksjni/comparator.cc

@@ -12,42 +12,33 @@
 #include <functional>
 #include <string>
 
-#include "include/org_rocksdb_Comparator.h"
-#include "include/org_rocksdb_DirectComparator.h"
+#include "include/org_rocksdb_AbstractComparator.h"
 #include "include/org_rocksdb_NativeComparatorWrapper.h"
 #include "rocksjni/comparatorjnicallback.h"
 #include "rocksjni/portal.h"
 
-// <editor-fold desc="org.rocksdb.Comparator>
-
 /*
- * Class:     org_rocksdb_Comparator
- * Method:    createNewComparator0
- * Signature: ()J
+ * Class:     org_rocksdb_AbstractComparator
+ * Method:    createNewComparator
+ * Signature: (J)J
  */
-jlong Java_org_rocksdb_Comparator_createNewComparator0(JNIEnv* env,
-                                                       jobject jobj,
-                                                       jlong copt_handle) {
+jlong Java_org_rocksdb_AbstractComparator_createNewComparator(
+    JNIEnv* env, jobject jcomparator, jlong copt_handle) {
   auto* copt =
       reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(copt_handle);
-  auto* c = new rocksdb::ComparatorJniCallback(env, jobj, copt);
+  auto* c = new rocksdb::ComparatorJniCallback(env, jcomparator, copt);
   return reinterpret_cast<jlong>(c);
 }
-// </editor-fold>
-
-// <editor-fold desc="org.rocksdb.DirectComparator>
 
 /*
- * Class:     org_rocksdb_DirectComparator
- * Method:    createNewDirectComparator0
- * Signature: ()J
+ * Class:     org_rocksdb_AbstractComparator
+ * Method:    usingDirectBuffers
+ * Signature: (J)Z
  */
-jlong Java_org_rocksdb_DirectComparator_createNewDirectComparator0(
-    JNIEnv* env, jobject jobj, jlong copt_handle) {
-  auto* copt =
-      reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(copt_handle);
-  auto* c = new rocksdb::DirectComparatorJniCallback(env, jobj, copt);
-  return reinterpret_cast<jlong>(c);
+jboolean Java_org_rocksdb_AbstractComparator_usingDirectBuffers(
+    JNIEnv*, jobject, jlong jhandle) {
+  auto* c = reinterpret_cast<rocksdb::ComparatorJniCallback*>(jhandle);
+  return static_cast<jboolean>(c->m_options->direct_buffer);
 }
 
 /*
@@ -60,4 +51,3 @@ void Java_org_rocksdb_NativeComparatorWrapper_disposeInternal(
   auto* comparator = reinterpret_cast<rocksdb::Comparator*>(jcomparator_handle);
   delete comparator;
 }
-// </editor-fold>