[RocksJava] Support Snapshots

Summary:
Snapshots integration into RocksJava. Added support for the following functionalities:

- getSnapshot
- releaseSnapshot
- ReadOptions support to set a Snapshot
- ReadOptions support to retrieve Snapshot
- SnapshotTest

Test Plan:
make rocksdbjava
make jtest

Differential Revision: https://reviews.facebook.net/D24801

java/Makefile

@@ -46,6 +46,7 @@ test: java
 	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.OptionsTest
 	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.ReadOnlyTest
 	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.ReadOptionsTest
+	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.SnapshotTest
 	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.StatisticsCollectorTest
 	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.ComparatorOptionsTest
 	java -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.ComparatorTest

java/org/rocksdb/ReadOptions.java

@@ -80,6 +80,39 @@ public class ReadOptions extends RocksObject {
   private native void setFillCache(
       long handle, boolean fillCache);
 
+  /**
+   * <p>If "snapshot" is non-nullptr, read as of the supplied snapshot
+   * (which must belong to the DB that is being read and which must
+   * not have been released).  If "snapshot" is nullptr, use an implicit
+   * snapshot of the state at the beginning of this read operation.</p>
+   * <p>Default: null</p>
+   *
+   * @param snapshot {@link Snapshot} instance
+   * @return the reference to the current ReadOptions.
+   */
+  public ReadOptions setSnapshot(Snapshot snapshot) {
+    assert(isInitialized());
+    setSnapshot(nativeHandle_, snapshot.nativeHandle_);
+    return this;
+  }
+  private native void setSnapshot(long handle, long snapshotHandle);
+
+  /**
+   * Returns the currently assigned Snapshot instance.
+   *
+   * @return the Snapshot assigned to this instance. If no Snapshot
+   *     is assigned null.
+   */
+  public Snapshot snapshot() {
+    assert(isInitialized());
+    long snapshotHandle = snapshot(nativeHandle_);
+    if (snapshotHandle != 0) {
+      return new Snapshot(snapshotHandle);
+    }
+    return null;
+  }
+  private native long snapshot(long handle);
+
   /**
    * Specify to create a tailing iterator -- a special iterator that has a
    * view of the complete database (i.e. it can also be used to read newly

java/org/rocksdb/RocksDB.java

@@ -891,6 +891,38 @@ public class RocksDB extends RocksObject {
     return new RocksIterator(iterator0(nativeHandle_));
   }
 
+
+  /**
+   * <p>Return a handle to the current DB state. Iterators created with
+   * this handle will all observe a stable snapshot of the current DB
+   * state. The caller must call ReleaseSnapshot(result) when the
+   * snapshot is no longer needed.</p>
+   *
+   * <p>nullptr will be returned if the DB fails to take a snapshot or does
+   * not support snapshot.</p>
+   *
+   * @return Snapshot
+   */
+  public Snapshot getSnapshot() {
+    long snapshotHandle = getSnapshot(nativeHandle_);
+    if (snapshotHandle != 0) {
+      return new Snapshot(snapshotHandle);
+    }
+    return null;
+  }
+
+  /**
+   * Release a previously acquired snapshot.  The caller must not
+   * use "snapshot" after this call.
+   *
+   * @param snapshot {@link Snapshot} instance
+   */
+  public void releaseSnapshot(final Snapshot snapshot) {
+    if (snapshot != null) {
+      releaseSnapshot(nativeHandle_, snapshot.nativeHandle_);
+    }
+  }
+
   /**
    * Return a heap-allocated iterator over the contents of the database.
    * The result of newIterator() is initially invalid (caller must
@@ -1052,6 +1084,9 @@ public class RocksDB extends RocksObject {
   protected native long iterator0(long handle, long cfHandle);
   protected native long[] iterators(long handle,
       List<ColumnFamilyHandle> columnFamilyNames) throws RocksDBException;
+  protected native long getSnapshot(long nativeHandle);
+  protected native void releaseSnapshot(
+      long nativeHandle, long snapshotHandle);
   private native void disposeInternal(long handle);
 
   private native long createColumnFamily(long handle, String name) throws RocksDBException;

java/org/rocksdb/Snapshot.java

@@ -0,0 +1,24 @@
+// Copyright (c) 2014, Facebook, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+
+package org.rocksdb;
+
+/**
+ * Snapshot of database
+ */
+public class Snapshot extends RocksObject {
+  Snapshot(long nativeHandle) {
+    super();
+    nativeHandle_ = nativeHandle;
+  }
+
+  /**
+   * Dont release C++ Snapshot pointer. The pointer
+   * to the snapshot is released by the database
+   * instance.
+   */
+  @Override protected void disposeInternal() {
+  }
+}

java/org/rocksdb/test/SnapshotTest.java

@@ -0,0 +1,82 @@
+// Copyright (c) 2014, Facebook, Inc.  All rights reserved.
+// This source code is licensed under the BSD-style license found in the
+// LICENSE file in the root directory of this source tree. An additional grant
+// of patent rights can be found in the PATENTS file in the same directory.
+package org.rocksdb.test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.Options;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.Snapshot;
+import org.rocksdb.WriteBatch;
+import org.rocksdb.WriteOptions;
+
+
+public class SnapshotTest
+{
+  static final String DB_PATH = "/tmp/rocksdbjni_snapshot_test";
+  static {
+    RocksDB.loadLibrary();
+  }
+
+  public static void main(String[] args){
+    RocksDB db = null;
+    Options options = new Options();
+    options.setCreateIfMissing(true);
+    try {
+      db = RocksDB.open(options, DB_PATH);
+      db.put("key".getBytes(), "value".getBytes());
+      // Get new Snapshot of database
+      Snapshot snapshot = db.getSnapshot();
+      ReadOptions readOptions = new ReadOptions();
+      // set snapshot in ReadOptions
+      readOptions.setSnapshot(snapshot);
+      // retrieve key value pair
+      assert(new String(db.get("key".getBytes()))
+          .equals("value"));
+      // retrieve key value pair created before
+      // the snapshot was made
+      assert(new String(db.get(readOptions,
+          "key".getBytes())).equals("value"));
+      // add new key/value pair
+      db.put("newkey".getBytes(), "newvalue".getBytes());
+      // using no snapshot the latest db entries
+      // will be taken into account
+      assert(new String(db.get("newkey".getBytes()))
+          .equals("newvalue"));
+      // snapshopot was created before newkey
+      assert(db.get(readOptions, "newkey".getBytes())
+          == null);
+      // Retrieve snapshot from read options
+      Snapshot sameSnapshot = readOptions.snapshot();
+      readOptions.setSnapshot(sameSnapshot);
+      // results must be the same with new Snapshot
+      // instance using the same native pointer
+      assert(new String(db.get(readOptions,
+          "key".getBytes())).equals("value"));
+      // update key value pair to newvalue
+      db.put("key".getBytes(), "newvalue".getBytes());
+      // read with previously created snapshot will
+      // read previous version of key value pair
+      assert(new String(db.get(readOptions,
+          "key".getBytes())).equals("value"));
+      // read for newkey using the snapshot must be
+      // null
+      assert(db.get(readOptions, "newkey".getBytes())
+          == null);
+      // release Snapshot
+      db.releaseSnapshot(snapshot);
+      // Close database
+      db.close();
+    }catch (RocksDBException e){
+      e.printStackTrace();
+      assert(false);
+    }
+    System.out.println("Passed SnapshotTest");
+  }
+}

java/rocksjni/options.cc

@@ -1774,8 +1774,6 @@ void Java_org_rocksdb_ReadOptions_setTailing(
       static_cast<bool>(jtailing);
 }
 
-/////////////////////////////////////////////////////////////////////
-// rocksdb::ComparatorOptions
 /*
  * Class:     org_rocksdb_ComparatorOptions
  * Method:    newComparatorOptions
@@ -1819,3 +1817,26 @@ void Java_org_rocksdb_ComparatorOptions_disposeInternal(
   delete reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle);
   rocksdb::ComparatorOptionsJni::setHandle(env, jobj, nullptr);
 }
+
+/*
+ * Class:     org_rocksdb_ReadOptions
+ * Method:    setSnapshot
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_ReadOptions_setSnapshot(
+    JNIEnv* env, jobject jobj, jlong jhandle, jlong jsnapshot) {
+  reinterpret_cast<rocksdb::ReadOptions*>(jhandle)->snapshot =
+      reinterpret_cast<rocksdb::Snapshot*>(jsnapshot);
+}
+
+/*
+ * Class:     org_rocksdb_ReadOptions
+ * Method:    snapshot
+ * Signature: (J)J
+ */
+jlong Java_org_rocksdb_ReadOptions_snapshot(
+    JNIEnv* env, jobject jobj, jlong jhandle) {
+  auto& snapshot =
+      reinterpret_cast<rocksdb::ReadOptions*>(jhandle)->snapshot;
+  return reinterpret_cast<jlong>(snapshot);
+}

java/rocksjni/rocksjni.cc

@@ -1031,6 +1031,28 @@ void Java_org_rocksdb_RocksDB_dropColumnFamily(
   }
 }
 
+/*
+ * Method:    getSnapshot
+ * Signature: (J)J
+ */
+jlong Java_org_rocksdb_RocksDB_getSnapshot(
+    JNIEnv* env, jobject jdb, jlong db_handle) {
+  auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
+  const rocksdb::Snapshot* snapshot = db->GetSnapshot();
+  return reinterpret_cast<jlong>(snapshot);
+}
+
+/*
+ * Method:    releaseSnapshot
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_RocksDB_releaseSnapshot(
+    JNIEnv* env, jobject jdb, jlong db_handle, jlong snapshot_handle) {
+  auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
+  auto snapshot = reinterpret_cast<rocksdb::Snapshot*>(snapshot_handle);
+  db->ReleaseSnapshot(snapshot);
+}
+
 /*
  * Class:     org_rocksdb_RocksDB
  * Method:    getProperty0