Provide byte[] version of SstFileWriter.merge to reduce GC Stall

Summary:
In Java API, `SstFileWriter.put/merge/delete` takes `Slice` type of key and value, which is a Java wrapper object around C++ Slice object.  The Slice object inherited [ `finalize`](https://github.com/facebook/rocksdb/blob/3c327ac2d0fd50bbd82fe1f1af5de909dad769e6/java/src/main/java/org/rocksdb/AbstractNativeReference.java#L69) method, which [added huge overhead](https://softwareengineering.stackexchange.com/questions/288715/is-overriding-object-finalize-really-bad/288753#288753) to JVM while creating new SstFile.

To address this issue, this PR overload the merge method to take Java byte array instead of the Slice object, and added unit test for it.

We also benchmark these two different merge function, where we could see GC Stall reduced from 50%  to 1%, and the throughput increased from 50MB to 200MB.
Closes https://github.com/facebook/rocksdb/pull/2746

Reviewed By: sagar0

Differential Revision: D5653145

Pulled By: scv119

fbshipit-source-id: b55ea58554b573d0b1c6f6170f8d9223811bc4f5

java/rocksjni/sst_file_writerjni.cc

@@ -77,9 +77,9 @@ void Java_org_rocksdb_SstFileWriter_open(JNIEnv *env, jobject jobj,
  * Method:    put
  * Signature: (JJJ)V
  */
-void Java_org_rocksdb_SstFileWriter_put(JNIEnv *env, jobject jobj,
-                                        jlong jhandle, jlong jkey_handle,
-                                        jlong jvalue_handle) {
+void Java_org_rocksdb_SstFileWriter_put__JJJ(JNIEnv *env, jobject jobj,
+                                             jlong jhandle, jlong jkey_handle,
+                                             jlong jvalue_handle) {
   auto *key_slice = reinterpret_cast<rocksdb::Slice *>(jkey_handle);
   auto *value_slice = reinterpret_cast<rocksdb::Slice *>(jvalue_handle);
   rocksdb::Status s =
@@ -90,14 +90,51 @@ void Java_org_rocksdb_SstFileWriter_put(JNIEnv *env, jobject jobj,
   }
 }
 
+/*
+ * Class:     org_rocksdb_SstFileWriter
+ * Method:    put
+ * Signature: (JJJ)V
+ */
+ void Java_org_rocksdb_SstFileWriter_put__J_3B_3B(JNIEnv *env, jobject jobj,
+                                                  jlong jhandle, jbyteArray jkey,
+                                                  jbyteArray jval) {
+  jbyte* key = env->GetByteArrayElements(jkey, nullptr);
+  if(key == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return;
+  }
+  rocksdb::Slice key_slice(
+      reinterpret_cast<char*>(key),  env->GetArrayLength(jkey));
+
+  jbyte* value = env->GetByteArrayElements(jval, nullptr);
+  if(value == nullptr) {
+    // exception thrown: OutOfMemoryError
+    env->ReleaseByteArrayElements(jkey, key, JNI_ABORT);
+    return;
+  }
+  rocksdb::Slice value_slice(
+      reinterpret_cast<char*>(value),  env->GetArrayLength(jval));
+
+  rocksdb::Status s =
+  reinterpret_cast<rocksdb::SstFileWriter *>(jhandle)->Put(key_slice,
+                                                           value_slice);
+
+  env->ReleaseByteArrayElements(jkey, key, JNI_ABORT);
+  env->ReleaseByteArrayElements(jval, value, JNI_ABORT);
+
+  if (!s.ok()) {
+    rocksdb::RocksDBExceptionJni::ThrowNew(env, s);
+  }
+}
+
 /*
  * Class:     org_rocksdb_SstFileWriter
  * Method:    merge
  * Signature: (JJJ)V
  */
-void Java_org_rocksdb_SstFileWriter_merge(JNIEnv *env, jobject jobj,
-                                          jlong jhandle, jlong jkey_handle,
-                                          jlong jvalue_handle) {
+void Java_org_rocksdb_SstFileWriter_merge__JJJ(JNIEnv *env, jobject jobj,
+                                               jlong jhandle, jlong jkey_handle,
+                                               jlong jvalue_handle) {
   auto *key_slice = reinterpret_cast<rocksdb::Slice *>(jkey_handle);
   auto *value_slice = reinterpret_cast<rocksdb::Slice *>(jvalue_handle);
   rocksdb::Status s =
@@ -108,13 +145,76 @@ void Java_org_rocksdb_SstFileWriter_merge(JNIEnv *env, jobject jobj,
   }
 }
 
+/*
+ * Class:     org_rocksdb_SstFileWriter
+ * Method:    merge
+ * Signature: (J[B[B)V
+ */
+void Java_org_rocksdb_SstFileWriter_merge__J_3B_3B(JNIEnv *env, jobject jobj,
+                                                   jlong jhandle, jbyteArray jkey,
+                                                   jbyteArray jval) {
+
+  jbyte* key = env->GetByteArrayElements(jkey, nullptr);
+  if(key == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return;
+  }
+  rocksdb::Slice key_slice(
+      reinterpret_cast<char*>(key),  env->GetArrayLength(jkey));
+
+  jbyte* value = env->GetByteArrayElements(jval, nullptr);
+  if(value == nullptr) {
+    // exception thrown: OutOfMemoryError
+    env->ReleaseByteArrayElements(jkey, key, JNI_ABORT);
+    return;
+  }
+  rocksdb::Slice value_slice(
+      reinterpret_cast<char*>(value),  env->GetArrayLength(jval));
+
+  rocksdb::Status s =
+    reinterpret_cast<rocksdb::SstFileWriter *>(jhandle)->Merge(key_slice,
+                                                               value_slice);
+
+  env->ReleaseByteArrayElements(jkey, key, JNI_ABORT);
+  env->ReleaseByteArrayElements(jval, value, JNI_ABORT);
+
+  if (!s.ok()) {
+    rocksdb::RocksDBExceptionJni::ThrowNew(env, s);
+  }
+}
+
+/*
+ * Class:     org_rocksdb_SstFileWriter
+ * Method:    delete
+ * Signature: (JJJ)V
+ */
+void Java_org_rocksdb_SstFileWriter_delete__J_3B(JNIEnv *env, jobject jobj,
+                                               jlong jhandle, jbyteArray jkey) {
+  jbyte* key = env->GetByteArrayElements(jkey, nullptr);
+  if(key == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return;
+  }
+  rocksdb::Slice key_slice(
+      reinterpret_cast<char*>(key),  env->GetArrayLength(jkey));
+
+  rocksdb::Status s =
+    reinterpret_cast<rocksdb::SstFileWriter *>(jhandle)->Delete(key_slice);
+
+  env->ReleaseByteArrayElements(jkey, key, JNI_ABORT);
+
+  if (!s.ok()) {
+    rocksdb::RocksDBExceptionJni::ThrowNew(env, s);
+  }
+}
+
 /*
  * Class:     org_rocksdb_SstFileWriter
  * Method:    delete
  * Signature: (JJJ)V
  */
-void Java_org_rocksdb_SstFileWriter_delete(JNIEnv *env, jobject jobj,
-                                           jlong jhandle, jlong jkey_handle) {
+ void Java_org_rocksdb_SstFileWriter_delete__JJ(JNIEnv *env, jobject jobj,
+  jlong jhandle, jlong jkey_handle) {
   auto *key_slice = reinterpret_cast<rocksdb::Slice *>(jkey_handle);
   rocksdb::Status s =
     reinterpret_cast<rocksdb::SstFileWriter *>(jhandle)->Delete(*key_slice);

java/src/main/java/org/rocksdb/SstFileWriter.java

@@ -117,6 +117,20 @@ public class SstFileWriter extends RocksObject {
     put(nativeHandle_, key.getNativeHandle(), value.getNativeHandle());
   }
 
+ /**
+   * Add a Put key with value to currently opened file.
+   *
+   * @param key the specified key to be inserted.
+   * @param value the value associated with the specified key.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+public void put(final byte[] key, final byte[] value)
+    throws RocksDBException {
+  put(nativeHandle_, key, value);
+}
+
   /**
    * Add a Merge key with value to currently opened file.
    *
@@ -132,6 +146,21 @@ public class SstFileWriter extends RocksObject {
     merge(nativeHandle_, key.getNativeHandle(), value.getNativeHandle());
   }
 
+  /**
+   * Add a Merge key with value to currently opened file.
+   *
+   * @param key the specified key to be merged.
+   * @param value the value to be merged with the current value for
+   * the specified key.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void merge(final byte[] key, final byte[] value)
+      throws RocksDBException {
+    merge(nativeHandle_, key, value);
+  }
+
   /**
    * Add a Merge key with value to currently opened file.
    *
@@ -171,6 +200,18 @@ public class SstFileWriter extends RocksObject {
     delete(nativeHandle_, key.getNativeHandle());
   }
 
+  /**
+   * Add a deletion key to currently opened file.
+   *
+   * @param key the specified key to be deleted.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void delete(final byte[] key) throws RocksDBException {
+    delete(nativeHandle_, key);
+  }
+
   /**
    * Finish the process and close the sst file.
    *
@@ -193,13 +234,22 @@ public class SstFileWriter extends RocksObject {
 
   private native void put(final long handle, final long keyHandle,
       final long valueHandle) throws RocksDBException;
+      
+  private native void put(final long handle, final byte[] key,
+      final byte[] value) throws RocksDBException;
 
   private native void merge(final long handle, final long keyHandle,
       final long valueHandle) throws RocksDBException;
 
+  private native void merge(final long handle, final byte[] key,
+      final byte[] value) throws RocksDBException;
+
   private native void delete(final long handle, final long keyHandle)
       throws RocksDBException;
 
+  private native void delete(final long handle, final byte[] key)
+      throws RocksDBException;
+
   private native void finish(final long handle) throws RocksDBException;
 
   @Override protected final native void disposeInternal(final long handle);

java/src/test/java/org/rocksdb/SstFileWriterTest.java

@@ -30,7 +30,7 @@ public class SstFileWriterTest {
 
   @Rule public TemporaryFolder parentFolder = new TemporaryFolder();
 
-  enum OpType { PUT, MERGE, DELETE }
+  enum OpType { PUT, PUT_BYTES, MERGE, MERGE_BYTES, DELETE, DELETE_BYTES}
 
   class KeyValueWithOp {
     KeyValueWithOp(String key, String value, OpType opType) {
@@ -79,16 +79,27 @@ public class SstFileWriterTest {
       for (KeyValueWithOp keyValue : keyValues) {
         Slice keySlice = new Slice(keyValue.getKey());
         Slice valueSlice = new Slice(keyValue.getValue());
+        byte[] keyBytes = keyValue.getKey().getBytes();
+        byte[] valueBytes = keyValue.getValue().getBytes();
         switch (keyValue.getOpType()) {
           case PUT:
             sstFileWriter.put(keySlice, valueSlice);
             break;
+          case PUT_BYTES:
+            sstFileWriter.put(keyBytes, valueBytes);
+            break;
           case MERGE:
             sstFileWriter.merge(keySlice, valueSlice);
             break;
+          case MERGE_BYTES:
+            sstFileWriter.merge(keyBytes, valueBytes);
+            break;
           case DELETE:
             sstFileWriter.delete(keySlice);
             break;
+          case DELETE_BYTES:
+            sstFileWriter.delete(keyBytes);
+            break;
           default:
             fail("Unsupported op type");
         }
@@ -142,8 +153,12 @@ public class SstFileWriterTest {
     final List<KeyValueWithOp> keyValues = new ArrayList<>();
     keyValues.add(new KeyValueWithOp("key1", "value1", OpType.PUT));
     keyValues.add(new KeyValueWithOp("key2", "value2", OpType.PUT));
-    keyValues.add(new KeyValueWithOp("key3", "value3", OpType.MERGE));
-    keyValues.add(new KeyValueWithOp("key4", "", OpType.DELETE));
+    keyValues.add(new KeyValueWithOp("key3", "value3", OpType.PUT_BYTES));
+    keyValues.add(new KeyValueWithOp("key4", "value4", OpType.MERGE));
+    keyValues.add(new KeyValueWithOp("key5", "value5", OpType.MERGE_BYTES));
+    keyValues.add(new KeyValueWithOp("key6", "", OpType.DELETE));
+    keyValues.add(new KeyValueWithOp("key7", "", OpType.DELETE));
+
 
     final File sstFile = newSstFile(keyValues, false);
     final File dbFolder = parentFolder.newFolder(DB_DIRECTORY_NAME);
@@ -161,7 +176,10 @@ public class SstFileWriterTest {
       assertThat(db.get("key1".getBytes())).isEqualTo("value1".getBytes());
       assertThat(db.get("key2".getBytes())).isEqualTo("value2".getBytes());
       assertThat(db.get("key3".getBytes())).isEqualTo("value3".getBytes());
-      assertThat(db.get("key4".getBytes())).isEqualTo(null);
+      assertThat(db.get("key4".getBytes())).isEqualTo("value4".getBytes());
+      assertThat(db.get("key5".getBytes())).isEqualTo("value5".getBytes());
+      assertThat(db.get("key6".getBytes())).isEqualTo(null);
+      assertThat(db.get("key7".getBytes())).isEqualTo(null);
     }
   }