Merge pull request #466 from fyrz/RocksJava-Support-ReadOptions-Iterator

[RocksJava] Allow using ReadOptions with iterators

java/org/rocksdb/RocksDB.java

@@ -1107,21 +1107,40 @@ public class RocksDB extends RocksObject {
   }
 
   /**
-   * Return a heap-allocated iterator over the contents of the database.
-   * The result of newIterator() is initially invalid (caller must
-   * call one of the Seek methods on the iterator before using it).
+   * <p>Return a heap-allocated iterator over the contents of the
+   * database. The result of newIterator() is initially invalid
+   * (caller must call one of the Seek methods on the iterator
+   * before using it).</p>
    *
-   * Caller should close the iterator when it is no longer needed.
+   * <p>Caller should close the iterator when it is no longer needed.
    * The returned iterator should be closed before this db is closed.
+   * </p>
    *
    * @return instance of iterator object.
    */
   public RocksIterator newIterator() {
-    return new RocksIterator(this, iterator0(nativeHandle_));
+    return new RocksIterator(this, iterator(nativeHandle_));
   }
 
-
   /**
+   * <p>Return a heap-allocated iterator over the contents of the
+   * database. The result of newIterator() is initially invalid
+   * (caller must call one of the Seek methods on the iterator
+   * before using it).</p>
+   *
+   * <p>Caller should close the iterator when it is no longer needed.
+   * The returned iterator should be closed before this db is closed.
+   * </p>
+   *
+   * @param readOptions {@link ReadOptions} instance.
+   * @return instance of iterator object.
+   */
+  public RocksIterator newIterator(ReadOptions readOptions) {
+    return new RocksIterator(this, iterator(nativeHandle_,
+        readOptions.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
@@ -1153,22 +1172,45 @@ public class RocksDB extends RocksObject {
   }
 
   /**
-   * Return a heap-allocated iterator over the contents of the database.
-   * The result of newIterator() is initially invalid (caller must
-   * call one of the Seek methods on the iterator before using it).
+   * <p>Return a heap-allocated iterator over the contents of the
+   * database. The result of newIterator() is initially invalid
+   * (caller must call one of the Seek methods on the iterator
+   * before using it).</p>
    *
-   * Caller should close the iterator when it is no longer needed.
+   * <p>Caller should close the iterator when it is no longer needed.
    * The returned iterator should be closed before this db is closed.
+   * </p>
    *
    * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
    *     instance
    * @return instance of iterator object.
    */
   public RocksIterator newIterator(ColumnFamilyHandle columnFamilyHandle) {
-    return new RocksIterator(this, iterator0(nativeHandle_,
+    return new RocksIterator(this, iteratorCF(nativeHandle_,
         columnFamilyHandle.nativeHandle_));
   }
 
+  /**
+   * <p>Return a heap-allocated iterator over the contents of the
+   * database. The result of newIterator() is initially invalid
+   * (caller must call one of the Seek methods on the iterator
+   * before using it).</p>
+   *
+   * <p>Caller should close the iterator when it is no longer needed.
+   * The returned iterator should be closed before this db is closed.
+   * </p>
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param readOptions {@link ReadOptions} instance.
+   * @return instance of iterator object.
+   */
+  public RocksIterator newIterator(ColumnFamilyHandle columnFamilyHandle,
+      ReadOptions readOptions) {
+    return new RocksIterator(this, iteratorCF(nativeHandle_,
+        columnFamilyHandle.nativeHandle_, readOptions.nativeHandle_));
+  }
+
   /**
    * Returns iterators from a consistent database state across multiple
    * column families. Iterators are heap allocated and need to be deleted
@@ -1184,10 +1226,31 @@ public class RocksDB extends RocksObject {
    */
   public List<RocksIterator> newIterators(
       List<ColumnFamilyHandle> columnFamilyHandleList) throws RocksDBException {
+    return newIterators(columnFamilyHandleList, new ReadOptions());
+  }
+
+  /**
+   * Returns iterators from a consistent database state across multiple
+   * column families. Iterators are heap allocated and need to be deleted
+   * before the db is deleted
+   *
+   * @param columnFamilyHandleList {@link java.util.List} containing
+   *     {@link org.rocksdb.ColumnFamilyHandle} instances.
+   * @param readOptions {@link ReadOptions} instance.
+   * @return {@link java.util.List} containing {@link org.rocksdb.RocksIterator}
+   *     instances
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public List<RocksIterator> newIterators(
+      List<ColumnFamilyHandle> columnFamilyHandleList,
+      ReadOptions readOptions) throws RocksDBException {
     List<RocksIterator> iterators =
         new ArrayList<>(columnFamilyHandleList.size());
 
-    long[] iteratorRefs = iterators(nativeHandle_, columnFamilyHandleList);
+    long[] iteratorRefs = iterators(nativeHandle_, columnFamilyHandleList,
+        readOptions.nativeHandle_);
     for (int i=0; i<columnFamilyHandleList.size(); i++){
       iterators.add(new RocksIterator(this, iteratorRefs[i]));
     }
@@ -1637,10 +1700,14 @@ public class RocksDB extends RocksObject {
       String property, int propertyLength) throws RocksDBException;
   protected native long getLongProperty(long nativeHandle, long cfHandle,
       String property, int propertyLength) throws RocksDBException;
-  protected native long iterator0(long handle);
-  protected native long iterator0(long handle, long cfHandle);
+  protected native long iterator(long handle);
+  protected native long iterator(long handle, long readOptHandle);
+  protected native long iteratorCF(long handle, long cfHandle);
+  protected native long iteratorCF(long handle, long cfHandle,
+      long readOptHandle);
   protected native long[] iterators(long handle,
-      List<ColumnFamilyHandle> columnFamilyNames) throws RocksDBException;
+      List<ColumnFamilyHandle> columnFamilyNames, long readOptHandle)
+      throws RocksDBException;
   protected native long getSnapshot(long nativeHandle);
   protected native void releaseSnapshot(
       long nativeHandle, long snapshotHandle);

java/org/rocksdb/test/SnapshotTest.java

@@ -8,11 +8,7 @@ import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
-import org.rocksdb.Options;
-import org.rocksdb.ReadOptions;
-import org.rocksdb.RocksDB;
-import org.rocksdb.RocksDBException;
-import org.rocksdb.Snapshot;
+import org.rocksdb.*;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -94,4 +90,127 @@ public class SnapshotTest {
       }
     }
   }
+
+  @Test
+  public void iteratorWithSnapshot() throws RocksDBException {
+    RocksDB db = null;
+    Options options = null;
+    ReadOptions readOptions = null;
+    RocksIterator iterator = null;
+    RocksIterator snapshotIterator = null;
+    try {
+
+      options = new Options();
+      options.setCreateIfMissing(true);
+
+      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath());
+      db.put("key".getBytes(), "value".getBytes());
+      // Get new Snapshot of database
+      Snapshot snapshot = db.getSnapshot();
+      readOptions = new ReadOptions();
+      // set snapshot in ReadOptions
+      readOptions.setSnapshot(snapshot);
+      db.put("key2".getBytes(), "value2".getBytes());
+
+      // iterate over current state of db
+      iterator = db.newIterator();
+      iterator.seekToFirst();
+      assertThat(iterator.isValid()).isTrue();
+      assertThat(iterator.key()).isEqualTo("key".getBytes());
+      iterator.next();
+      assertThat(iterator.isValid()).isTrue();
+      assertThat(iterator.key()).isEqualTo("key2".getBytes());
+      iterator.next();
+      assertThat(iterator.isValid()).isFalse();
+
+      // iterate using a snapshot
+      snapshotIterator = db.newIterator(readOptions);
+      snapshotIterator.seekToFirst();
+      assertThat(snapshotIterator.isValid()).isTrue();
+      assertThat(snapshotIterator.key()).isEqualTo("key".getBytes());
+      snapshotIterator.next();
+      assertThat(snapshotIterator.isValid()).isFalse();
+
+      // release Snapshot
+      db.releaseSnapshot(snapshot);
+    } finally {
+      if (iterator != null) {
+        iterator.dispose();
+      }
+      if (snapshotIterator != null) {
+        snapshotIterator.dispose();
+      }
+      if (db != null) {
+        db.close();
+      }
+      if (options != null) {
+        options.dispose();
+      }
+      if (readOptions != null) {
+        readOptions.dispose();
+      }
+    }
+  }
+
+  @Test
+  public void iteratorWithSnapshotOnColumnFamily() throws RocksDBException {
+    RocksDB db = null;
+    Options options = null;
+    ReadOptions readOptions = null;
+    RocksIterator iterator = null;
+    RocksIterator snapshotIterator = null;
+    try {
+
+      options = new Options();
+      options.setCreateIfMissing(true);
+
+      db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath());
+      db.put("key".getBytes(), "value".getBytes());
+      // Get new Snapshot of database
+      Snapshot snapshot = db.getSnapshot();
+      readOptions = new ReadOptions();
+      // set snapshot in ReadOptions
+      readOptions.setSnapshot(snapshot);
+      db.put("key2".getBytes(), "value2".getBytes());
+
+      // iterate over current state of column family
+      iterator = db.newIterator(db.getDefaultColumnFamily());
+      iterator.seekToFirst();
+      assertThat(iterator.isValid()).isTrue();
+      assertThat(iterator.key()).isEqualTo("key".getBytes());
+      iterator.next();
+      assertThat(iterator.isValid()).isTrue();
+      assertThat(iterator.key()).isEqualTo("key2".getBytes());
+      iterator.next();
+      assertThat(iterator.isValid()).isFalse();
+
+      // iterate using a snapshot on default column family
+      snapshotIterator = db.newIterator(db.getDefaultColumnFamily(),
+          readOptions);
+      snapshotIterator.seekToFirst();
+      assertThat(snapshotIterator.isValid()).isTrue();
+      assertThat(snapshotIterator.key()).isEqualTo("key".getBytes());
+      snapshotIterator.next();
+      assertThat(snapshotIterator.isValid()).isFalse();
+
+      // release Snapshot
+      db.releaseSnapshot(snapshot);
+    } finally {
+      if (iterator != null) {
+        iterator.dispose();
+      }
+      if (snapshotIterator != null) {
+        snapshotIterator.dispose();
+      }
+      if (db != null) {
+        db.close();
+      }
+      if (options != null) {
+        options.dispose();
+      }
+      if (readOptions != null) {
+        readOptions.dispose();
+      }
+    }
+  }
 }

java/rocksjni/rocksjni.cc

@@ -1142,40 +1142,85 @@ void Java_org_rocksdb_RocksDB_disposeInternal(
   delete reinterpret_cast<rocksdb::DB*>(jhandle);
 }
 
-/*
- * Class:     org_rocksdb_RocksDB
- * Method:    iterator0
- * Signature: (J)J
- */
-jlong Java_org_rocksdb_RocksDB_iterator0__J(
-    JNIEnv* env, jobject jdb, jlong db_handle) {
-  auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
-  rocksdb::Iterator* iterator = db->NewIterator(rocksdb::ReadOptions());
+jlong rocksdb_iterator_helper(
+    rocksdb::DB* db, rocksdb::ReadOptions read_options,
+    rocksdb::ColumnFamilyHandle* cf_handle) {
+  rocksdb::Iterator* iterator = nullptr;
+  if (cf_handle != nullptr) {
+    iterator = db->NewIterator(read_options, cf_handle);
+  } else {
+    iterator = db->NewIterator(read_options);
+  }
   return reinterpret_cast<jlong>(iterator);
 }
 
 /*
  * Class:     org_rocksdb_RocksDB
- * Method:    iterator0
+ * Method:    iterator
+ * Signature: (J)J
+ */
+jlong Java_org_rocksdb_RocksDB_iterator__J(
+    JNIEnv* env, jobject jdb, jlong db_handle) {
+  auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
+  return rocksdb_iterator_helper(db, rocksdb::ReadOptions(),
+      nullptr);
+}
+
+/*
+ * Class:     org_rocksdb_RocksDB
+ * Method:    iterator
  * Signature: (JJ)J
  */
-jlong Java_org_rocksdb_RocksDB_iterator0__JJ(
+jlong Java_org_rocksdb_RocksDB_iterator__JJ(
+    JNIEnv* env, jobject jdb, jlong db_handle,
+    jlong jread_options_handle) {
+  auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
+  auto& read_options = *reinterpret_cast<rocksdb::ReadOptions*>(
+      jread_options_handle);
+  return rocksdb_iterator_helper(db, read_options,
+      nullptr);
+}
+
+/*
+ * Class:     org_rocksdb_RocksDB
+ * Method:    iteratorCF
+ * Signature: (JJ)J
+ */
+jlong Java_org_rocksdb_RocksDB_iteratorCF__JJ(
     JNIEnv* env, jobject jdb, jlong db_handle, jlong jcf_handle) {
   auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
   auto cf_handle = reinterpret_cast<rocksdb::ColumnFamilyHandle*>(jcf_handle);
-  rocksdb::Iterator* iterator = db->NewIterator(rocksdb::ReadOptions(),
-      cf_handle);
-  return reinterpret_cast<jlong>(iterator);
+  return rocksdb_iterator_helper(db, rocksdb::ReadOptions(),
+        cf_handle);
+}
+
+/*
+ * Class:     org_rocksdb_RocksDB
+ * Method:    iteratorCF
+ * Signature: (JJJ)J
+ */
+jlong Java_org_rocksdb_RocksDB_iteratorCF__JJJ(
+    JNIEnv* env, jobject jdb, jlong db_handle, jlong jcf_handle,
+    jlong jread_options_handle) {
+  auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
+  auto cf_handle = reinterpret_cast<rocksdb::ColumnFamilyHandle*>(jcf_handle);
+  auto& read_options = *reinterpret_cast<rocksdb::ReadOptions*>(
+      jread_options_handle);
+  return rocksdb_iterator_helper(db, read_options,
+        cf_handle);
 }
 
 /*
  * Class:     org_rocksdb_RocksDB
  * Method:    iterators
- * Signature: (JLjava/util/List;)[J
+ * Signature: (JLjava/util/List;J)[J
  */
 jlongArray Java_org_rocksdb_RocksDB_iterators(
-    JNIEnv* env, jobject jdb, jlong db_handle, jobject jcfhandle_list) {
+    JNIEnv* env, jobject jdb, jlong db_handle, jobject jcfhandle_list,
+    jlong jread_options_handle) {
   auto db = reinterpret_cast<rocksdb::DB*>(db_handle);
+  auto& read_options = *reinterpret_cast<rocksdb::ReadOptions*>(
+        jread_options_handle);
   std::vector<rocksdb::ColumnFamilyHandle*> cf_handles;
   std::vector<rocksdb::Iterator*> iterators;
 
@@ -1195,7 +1240,7 @@ jlongArray Java_org_rocksdb_RocksDB_iterators(
     }
   }
 
-  rocksdb::Status s = db->NewIterators(rocksdb::ReadOptions(),
+  rocksdb::Status s = db->NewIterators(read_options,
       cf_handles, &iterators);
   if (s.ok()) {
     jlongArray jLongArray =