Add iterator's SeekForPrev functionality to the java-api

Summary:
As discussed in #2742 , this pull-requests brings the iterator's [SeekForPrev()](https://github.com/facebook/rocksdb/wiki/SeekForPrev) functionality to the java-api. It affects all locations in the code where previously only Seek() was supported.

All code changes are essentially a copy & paste of the already existing implementations for Seek().
**Please Note**: the changes to the C++ code were applied without fully understanding its effect, so please take a closer look. However, since Seek() and SeekForPrev() provide exactly the same signature, I do not expect any mistake here.

The java-tests are extended by new tests for the additional functionality.

Compilation (`make rocksdbjavastatic`) and test (`java/make test`) run without errors.
Closes https://github.com/facebook/rocksdb/pull/2747

Differential Revision: D5721011

Pulled By: sagar0

fbshipit-source-id: c1f951cddc321592c70dd2d32bc04892f3f119f8

java/rocksjni/iterator.cc

@@ -99,6 +99,29 @@ void Java_org_rocksdb_RocksIterator_seek0(
   env->ReleaseByteArrayElements(jtarget, target, JNI_ABORT);
 }
 
+/*
+ * Class:     org_rocksdb_RocksIterator
+ * Method:    seekForPrev0
+ * Signature: (J[BI)V
+ */
+void Java_org_rocksdb_RocksIterator_seekForPrev0(
+    JNIEnv* env, jobject jobj, jlong handle,
+    jbyteArray jtarget, jint jtarget_len) {
+  jbyte* target = env->GetByteArrayElements(jtarget, nullptr);
+  if(target == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return;
+  }
+
+  rocksdb::Slice target_slice(
+      reinterpret_cast<char*>(target), jtarget_len);
+
+  auto* it = reinterpret_cast<rocksdb::Iterator*>(handle);
+  it->SeekForPrev(target_slice);
+
+  env->ReleaseByteArrayElements(jtarget, target, JNI_ABORT);
+}
+
 /*
  * Class:     org_rocksdb_RocksIterator
  * Method:    status0

java/rocksjni/write_batch_with_index.cc

@@ -491,6 +491,29 @@ void Java_org_rocksdb_WBWIRocksIterator_seek0(
   env->ReleaseByteArrayElements(jtarget, target, JNI_ABORT);
 }
 
+/*
+ * Class:     org_rocksdb_WBWIRocksIterator
+ * Method:    seekForPrev0
+ * Signature: (J[BI)V
+ */
+void Java_org_rocksdb_WBWIRocksIterator_seekForPrev0(
+    JNIEnv* env, jobject jobj, jlong handle, jbyteArray jtarget,
+    jint jtarget_len) {
+  auto* it = reinterpret_cast<rocksdb::WBWIIterator*>(handle);
+  jbyte* target = env->GetByteArrayElements(jtarget, nullptr);
+  if(target == nullptr) {
+    // exception thrown: OutOfMemoryError
+    return;
+  }
+
+  rocksdb::Slice target_slice(
+      reinterpret_cast<char*>(target), jtarget_len);
+
+  it->SeekForPrev(target_slice);
+
+  env->ReleaseByteArrayElements(jtarget, target, JNI_ABORT);
+}
+
 /*
  * Class:     org_rocksdb_WBWIRocksIterator
  * Method:    status0

java/src/main/java/org/rocksdb/AbstractRocksIterator.java

@@ -58,6 +58,12 @@ public abstract class AbstractRocksIterator<P extends RocksObject>
     seek0(nativeHandle_, target, target.length);
   }
 
+ @Override
+ public void seekForPrev(byte[] target) {
+   assert (isOwningHandle());
+   seekForPrev0(nativeHandle_, target, target.length);
+ }
+
   @Override
   public void next() {
     assert (isOwningHandle());
@@ -97,5 +103,6 @@ public abstract class AbstractRocksIterator<P extends RocksObject>
   abstract void next0(long handle);
   abstract void prev0(long handle);
   abstract void seek0(long handle, byte[] target, int targetLen);
+  abstract void seekForPrev0(long handle, byte[] target, int targetLen);
   abstract void status0(long handle) throws RocksDBException;
 }

java/src/main/java/org/rocksdb/RocksIterator.java

@@ -57,6 +57,7 @@ public class RocksIterator extends AbstractRocksIterator<RocksDB> {
   @Override final native void next0(long handle);
   @Override final native void prev0(long handle);
   @Override final native void seek0(long handle, byte[] target, int targetLen);
+  @Override final native void seekForPrev0(long handle, byte[] target, int targetLen);
   @Override final native void status0(long handle) throws RocksDBException;
 
   private native byte[] key0(long handle);

java/src/main/java/org/rocksdb/RocksIteratorInterface.java

@@ -52,6 +52,18 @@ public interface RocksIteratorInterface {
    */
   void seek(byte[] target);
 
+  /**
+   * <p>Position at the first entry in the source whose key is that or
+   * before target.</p>
+   *
+   * <p>The iterator is valid after this call if the source contains
+   * a key that comes at or before target.</p>
+   *
+   * @param target byte array describing a key or a
+   *               key prefix to seek for.
+   */
+  void seekForPrev(byte[] target);
+
   /**
    * <p>Moves to the next entry in the source.  After this call, Valid() is
    * true if the iterator was not positioned at the last entry in the source.</p>

java/src/main/java/org/rocksdb/WBWIRocksIterator.java

@@ -45,6 +45,7 @@ public class WBWIRocksIterator
   @Override final native void next0(long handle);
   @Override final native void prev0(long handle);
   @Override final native void seek0(long handle, byte[] target, int targetLen);
+  @Override final native void seekForPrev0(long handle, byte[] target, int targetLen);
   @Override final native void status0(long handle) throws RocksDBException;
 
   private native long[] entry1(final long handle);

java/src/test/java/org/rocksdb/RocksIteratorTest.java

@@ -53,6 +53,48 @@ public class RocksIteratorTest {
         assertThat(iterator.value()).isEqualTo("value2".getBytes());
         iterator.status();
       }
+
+      try (final RocksIterator iterator = db.newIterator()) {
+        iterator.seek("key0".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key1".getBytes());
+
+        iterator.seek("key1".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key1".getBytes());
+
+        iterator.seek("key1.5".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key2".getBytes());
+
+        iterator.seek("key2".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key2".getBytes());
+
+        iterator.seek("key3".getBytes());
+        assertThat(iterator.isValid()).isFalse();
+      }
+
+      try (final RocksIterator iterator = db.newIterator()) {
+        iterator.seekForPrev("key0".getBytes());
+        assertThat(iterator.isValid()).isFalse();
+
+        iterator.seekForPrev("key1".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key1".getBytes());
+
+        iterator.seekForPrev("key1.5".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key1".getBytes());
+
+        iterator.seekForPrev("key2".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key2".getBytes());
+
+        iterator.seekForPrev("key3".getBytes());
+        assertThat(iterator.isValid()).isTrue();
+        assertThat(iterator.key()).isEqualTo("key2".getBytes());
+      }
     }
   }
 }

java/src/test/java/org/rocksdb/util/BytewiseComparatorTest.java

@@ -27,6 +27,9 @@ import static org.junit.Assert.*;
  */
 public class BytewiseComparatorTest {
 
+  private List<String> source_strings = Arrays.asList("b", "d", "f", "h", "j", "l");
+  private List<String> interleaving_strings = Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m");
+
   /**
    * Open the database using the C++ BytewiseComparatorImpl
    * and test the results against our Java BytewiseComparator
@@ -42,7 +45,6 @@ public class BytewiseComparatorTest {
         doRandomIterationTest(
             db,
             toJavaComparator(new BytewiseComparator(new ComparatorOptions())),
-            Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i"),
             rnd,
             8, 100, 3
         );
@@ -67,7 +69,6 @@ public class BytewiseComparatorTest {
         doRandomIterationTest(
             db,
             toJavaComparator(new BytewiseComparator(new ComparatorOptions())),
-            Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i"),
             rnd,
             8, 100, 3
         );
@@ -94,7 +95,6 @@ public class BytewiseComparatorTest {
             toJavaComparator(new DirectBytewiseComparator(
                 new ComparatorOptions())
             ),
-            Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i"),
             rnd,
             8, 100, 3
         );
@@ -121,7 +121,6 @@ public class BytewiseComparatorTest {
             toJavaComparator(new DirectBytewiseComparator(
                 new ComparatorOptions())
             ),
-            Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i"),
             rnd,
             8, 100, 3
         );
@@ -148,7 +147,6 @@ public class BytewiseComparatorTest {
             toJavaComparator(
                 new ReverseBytewiseComparator(new ComparatorOptions())
             ),
-            Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i"),
             rnd,
             8, 100, 3
         );
@@ -176,7 +174,6 @@ public class BytewiseComparatorTest {
             toJavaComparator(
                 new ReverseBytewiseComparator(new ComparatorOptions())
             ),
-            Arrays.asList("a", "b", "c", "d", "e", "f", "g", "h", "i"),
             rnd,
             8, 100, 3
         );
@@ -188,7 +185,7 @@ public class BytewiseComparatorTest {
 
   private void doRandomIterationTest(
       final RocksDB db, final java.util.Comparator<String> javaComparator,
-      final List<String> source_strings, final Random rnd,
+      final Random rnd,
       final int num_writes, final int num_iter_ops,
       final int num_trigger_flush) throws RocksDBException {
 
@@ -228,7 +225,7 @@ public class BytewiseComparatorTest {
       for (int i = 0; i < num_iter_ops; i++) {
         // Random walk and make sure iter and result_iter returns the
         // same key and value
-        final int type = rnd.nextInt(6);
+        final int type = rnd.nextInt(7);
         iter.status();
         switch (type) {
           case 0:
@@ -242,14 +239,22 @@ public class BytewiseComparatorTest {
             result_iter.seekToLast();
             break;
           case 2: {
-            // Seek to random key
-            final int key_idx = rnd.nextInt(source_strings.size());
-            final String key = source_strings.get(key_idx);
+            // Seek to random (existing or non-existing) key
+            final int key_idx = rnd.nextInt(interleaving_strings.size());
+            final String key = interleaving_strings.get(key_idx);
             iter.seek(bytes(key));
             result_iter.seek(bytes(key));
             break;
           }
-          case 3:
+          case 3: {
+            // SeekForPrev to random (existing or non-existing) key
+            final int key_idx = rnd.nextInt(interleaving_strings.size());
+            final String key = interleaving_strings.get(key_idx);
+            iter.seekForPrev(bytes(key));
+            result_iter.seekForPrev(bytes(key));
+            break;
+          }
+          case 4:
             // Next
             if (is_valid) {
               iter.next();
@@ -258,7 +263,7 @@ public class BytewiseComparatorTest {
               continue;
             }
             break;
-          case 4:
+          case 5:
             // Prev
             if (is_valid) {
               iter.prev();
@@ -268,7 +273,7 @@ public class BytewiseComparatorTest {
             }
             break;
           default: {
-            assert (type == 5);
+            assert (type == 6);
             final int key_idx = rnd.nextInt(source_strings.size());
             final String key = source_strings.get(key_idx);
             final byte[] result = db.get(new ReadOptions(), bytes(key));
@@ -413,6 +418,16 @@ public class BytewiseComparatorTest {
       }
     }
 
+    @Override
+    public void seekForPrev(final byte[] target) {
+      for(offset = entries.size()-1; offset >= 0; offset--) {
+        if(comparator.compare(entries.get(offset).getKey(),
+            (K)new String(target, StandardCharsets.UTF_8)) <= 0) {
+          return;
+        }
+      }
+    }
+
     /**
      * Is `a` a prefix of `b`
      *