Built-in comparator(s) in RocksJava

Extended Built-in comparators with ReverseBytewiseComparator.

Reverse key handling is under certain conditions essential. E.g. while
using timestamp versioned data.

As native-comparators were not available using JAVA-API. Both built-in comparators
were exposed via JNI to be set upon database creation time.

include/rocksdb/comparator.h

@@ -62,6 +62,10 @@ class Comparator {
 // must not be deleted.
 extern const Comparator* BytewiseComparator();
 
+// Return a builtin comparator that uses reverse lexicographic byte-wise
+// ordering. 
+extern const Comparator* ReverseBytewiseComparator();
+
 }  // namespace rocksdb
 
 #endif  // STORAGE_ROCKSDB_INCLUDE_COMPARATOR_H_

java/org/rocksdb/Options.java

@@ -18,6 +18,14 @@ public class Options extends RocksObject {
   }
   static final long DEFAULT_CACHE_SIZE = 8 << 20;
   static final int DEFAULT_NUM_SHARD_BITS = -1;
+
+  /**
+   * Builtin RocksDB comparators 
+   */
+  public enum BuiltinComparator {
+      BYTEWISE_COMPARATOR, REVERSE_BYTEWISE_COMPARATOR;
+  }
+
   /**
    * Construct options for opening a RocksDB.
    *
@@ -78,6 +86,21 @@ public class Options extends RocksObject {
     return createIfMissing(nativeHandle_);
   }
 
+  /**
+   * Set BuiltinComparator to be used with RocksDB. 
+   *
+   * Note: Comparator can be set once upon database creation.
+   *
+   * Default: BytewiseComparator.
+   * @param builtinComparator a BuiltinComparator type.
+   */
+  public void setBuiltinComparator(BuiltinComparator builtinComparator) {
+    assert(isInitialized());
+    setBuiltinComparator(nativeHandle_, builtinComparator.ordinal());
+  }
+
+  private native void setBuiltinComparator(long handle, int builtinComparator);
+
   /**
    * Amount of data to build up in memory (backed by an unsorted log
    * on disk) before converting to a sorted on-disk file.

java/rocksjni/options.cc

@@ -22,6 +22,7 @@
 #include "rocksdb/table.h"
 #include "rocksdb/slice_transform.h"
 #include "rocksdb/rate_limiter.h"
+#include "rocksdb/comparator.h"
 
 /*
  * Class:     org_rocksdb_Options
@@ -63,6 +64,23 @@ jboolean Java_org_rocksdb_Options_createIfMissing(
   return reinterpret_cast<rocksdb::Options*>(jhandle)->create_if_missing;
 }
 
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    useReverseBytewiseComparator
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_Options_setBuiltinComparator(
+    JNIEnv* env, jobject jobj, jlong jhandle, jint builtinComparator) {
+  switch (builtinComparator){
+    case 1:
+    	reinterpret_cast<rocksdb::Options*>(jhandle)->comparator = rocksdb::ReverseBytewiseComparator();
+    	break;
+    default:
+    	reinterpret_cast<rocksdb::Options*>(jhandle)->comparator = rocksdb::BytewiseComparator();
+	break;
+  }
+}
+
 /*
  * Class:     org_rocksdb_Options
  * Method:    setWriteBufferSize

util/comparator.cc

@@ -69,13 +69,29 @@ class BytewiseComparatorImpl : public Comparator {
     // *key is a run of 0xffs.  Leave it alone.
   }
 };
-}  // namespace
+
+class ReverseBytewiseComparatorImpl : public BytewiseComparatorImpl {
+ public:
+  ReverseBytewiseComparatorImpl() { }
+
+  virtual const char* Name() const {
+    return "leveldb.ReverseBytewiseComparator";
+  }
+
+  virtual int Compare(const Slice& a, const Slice& b) const {
+    return -a.compare(b);
+  }
+};
+
+}// namespace
 
 static port::OnceType once = LEVELDB_ONCE_INIT;
 static const Comparator* bytewise;
+static const Comparator* rbytewise;
 
 static void InitModule() {
   bytewise = new BytewiseComparatorImpl;
+  rbytewise= new ReverseBytewiseComparatorImpl;
 }
 
 const Comparator* BytewiseComparator() {
@@ -83,4 +99,9 @@ const Comparator* BytewiseComparator() {
   return bytewise;
 }
 
+const Comparator* ReverseBytewiseComparator() {
+  port::InitOnce(&once, InitModule);
+  return rbytewise;
+}
+
 }  // namespace rocksdb