diff --git a/.gitignore b/.gitignore
index 4ed05c5e3..ccbb46b03 100644
--- a/.gitignore
+++ b/.gitignore
@@ -31,8 +31,14 @@ coverage/COVERAGE_REPORT
 package/
 .phutil_module_cache
 tags
+
+java/out
 java/*.log
 java/include/org_rocksdb_*.h
+
+.idea/
+*.iml
+
 unity.cc
 java/crossbuild/.vagrant
 .vagrant/
diff --git a/java/Makefile b/java/Makefile
index ef8ccbae4..697df5175 100644
--- a/java/Makefile
+++ b/java/Makefile
@@ -1,4 +1,4 @@
-NATIVE_JAVA_CLASSES = org.rocksdb.RocksDB org.rocksdb.Options org.rocksdb.WriteBatch org.rocksdb.WriteBatchInternal org.rocksdb.WriteBatchTest org.rocksdb.WriteOptions org.rocksdb.BackupableDB org.rocksdb.BackupableDBOptions org.rocksdb.Statistics org.rocksdb.RocksIterator org.rocksdb.VectorMemTableConfig org.rocksdb.SkipListMemTableConfig org.rocksdb.HashLinkedListMemTableConfig org.rocksdb.HashSkipListMemTableConfig org.rocksdb.PlainTableConfig org.rocksdb.BlockBasedTableConfig org.rocksdb.ReadOptions org.rocksdb.Filter org.rocksdb.BloomFilter org.rocksdb.RestoreOptions org.rocksdb.RestoreBackupableDB org.rocksdb.RocksEnv org.rocksdb.GenericRateLimiterConfig org.rocksdb.ColumnFamilyHandle
+NATIVE_JAVA_CLASSES = org.rocksdb.RocksDB org.rocksdb.Options org.rocksdb.WriteBatch org.rocksdb.WriteBatchInternal org.rocksdb.WriteBatchTest org.rocksdb.WriteOptions org.rocksdb.BackupableDB org.rocksdb.BackupableDBOptions org.rocksdb.Statistics org.rocksdb.RocksIterator org.rocksdb.VectorMemTableConfig org.rocksdb.SkipListMemTableConfig org.rocksdb.HashLinkedListMemTableConfig org.rocksdb.HashSkipListMemTableConfig org.rocksdb.PlainTableConfig org.rocksdb.BlockBasedTableConfig org.rocksdb.ReadOptions org.rocksdb.Filter org.rocksdb.BloomFilter org.rocksdb.ComparatorOptions org.rocksdb.AbstractComparator org.rocksdb.Comparator org.rocksdb.DirectComparator org.rocksdb.AbstractSlice org.rocksdb.Slice org.rocksdb.DirectSlice org.rocksdb.RestoreOptions org.rocksdb.RestoreBackupableDB org.rocksdb.RocksEnv org.rocksdb.GenericRateLimiterConfig org.rocksdb.ColumnFamilyHandle
 
 ROCKSDB_MAJOR = $(shell egrep "ROCKSDB_MAJOR.[0-9]" ../include/rocksdb/version.h | cut -d ' ' -f 3)
 ROCKSDB_MINOR = $(shell egrep "ROCKSDB_MINOR.[0-9]" ../include/rocksdb/version.h | cut -d ' ' -f 3)
@@ -46,7 +46,9 @@ test: java
 	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.StatisticsCollectorTest
-	@rm -rf /tmp/rocksdbjni_*
+	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 -ea -Djava.library.path=.:../ -cp "$(ROCKSDB_JAR):.:./*" org.rocksdb.test.DirectComparatorTest
 
 db_bench: java
 	javac org/rocksdb/benchmark/*.java
diff --git a/java/org/rocksdb/AbstractComparator.java b/java/org/rocksdb/AbstractComparator.java
new file mode 100644
index 000000000..8de50e271
--- /dev/null
+++ b/java/org/rocksdb/AbstractComparator.java
@@ -0,0 +1,95 @@
+// 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;
+
+/**
+ * Comparators are used by RocksDB to determine
+ * the ordering of keys.
+ *
+ * This class is package private, implementers
+ * should extend either of the public abstract classes:
+ *   @see org.rocksdb.Comparator
+ *   @see org.rocksdb.DirectComparator
+ */
+public abstract class AbstractComparator<T extends AbstractSlice>
+    extends RocksObject {
+
+  /**
+   * The name of the comparator.  Used to check for comparator
+   * mismatches (i.e., a DB created with one comparator is
+   * accessed using a different comparator).
+   *
+   * A new name should be used whenever
+   * the comparator implementation changes in a way that will cause
+   * the relative ordering of any two keys to change.
+   *
+   * Names starting with "rocksdb." are reserved and should not be used.
+   *
+   * @return The name of this comparator implementation
+   */
+  public abstract String name();
+
+  /**
+   * Three-way key comparison
+   *
+   *  @param a Slice access to first key
+   *  @param b Slice access to second key
+   *
+   *  @return Should return either:
+   *    1) < 0 if "a" < "b"
+   *    2) == 0 if "a" == "b"
+   *    3) > 0 if "a" > "b"
+   */
+  public abstract int compare(final T a, final T b);
+
+  /**
+   * Used to reduce the space requirements
+   * for internal data structures like index blocks.
+   *
+   * If start < limit, you may return a new start which is a
+   * shorter string in [start, limit).
+   *
+   * Simple comparator implementations may return null if they
+   * wish to use start unchanged. i.e., an implementation of
+   * this method that does nothing is correct.
+   *
+   * @return a shorter start, or null
+   */
+  public String findShortestSeparator(final String start, final T limit) {
+      return null;
+  }
+
+  /**
+   * Used to reduce the space requirements
+   * for internal data structures like index blocks.
+   *
+   * You may return a new short key (key1) where
+   * key1 >= key.
+   *
+   * Simple comparator implementations may return null if they
+   * wish to leave the key unchanged. i.e., an implementation of
+   * this method that does nothing is correct.
+   *
+   * @return a shorter key, or null
+   */
+  public String findShortSuccessor(final String key) {
+      return null;
+  }
+
+  /**
+   * Deletes underlying C++ comparator pointer.
+   *
+   * Note that this function should be called only after all
+   * RocksDB instances referencing the comparator are closed.
+   * Otherwise an undefined behavior will occur.
+   */
+  @Override protected void disposeInternal() {
+    assert(isInitialized());
+    disposeInternal(nativeHandle_);
+  }
+
+  private native void disposeInternal(long handle);
+}
diff --git a/java/org/rocksdb/AbstractSlice.java b/java/org/rocksdb/AbstractSlice.java
new file mode 100644
index 000000000..971bd7c1a
--- /dev/null
+++ b/java/org/rocksdb/AbstractSlice.java
@@ -0,0 +1,167 @@
+// 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;
+
+/**
+ * Slices are used by RocksDB to provide
+ * efficient access to keys and values.
+ *
+ * This class is package private, implementers
+ * should extend either of the public abstract classes:
+ *   @see org.rocksdb.Slice
+ *   @see org.rocksdb.DirectSlice
+ *
+ * Regards the lifecycle of Java Slices in RocksDB:
+ *   At present when you configure a Comparator from Java, it creates an
+ *   instance of a C++ BaseComparatorJniCallback subclass and
+ *   passes that to RocksDB as the comparator. That subclass of
+ *   BaseComparatorJniCallback creates the Java
+ *   {@see org.rocksdb.AbstractSlice} subclass Objects. When you dispose
+ *   the Java {@see org.rocksdb.AbstractComparator} subclass, it disposes the
+ *   C++ BaseComparatorJniCallback subclass, which in turn destroys the
+ *   Java {@see org.rocksdb.AbstractSlice} subclass Objects.
+ */
+abstract class AbstractSlice<T> extends RocksObject {
+
+  /**
+   * Returns the data of the slice.
+   *
+   * @return The slice data. Note, the type of access is
+   *   determined by the subclass
+   *   @see org.rocksdb.AbstractSlice#data0(long).
+   */
+  public T data() {
+    assert (isInitialized());
+    return data0(nativeHandle_);
+  }
+
+  /**
+   * Access to the data is provided by the
+   * subtype as it needs to handle the
+   * generic typing.
+   *
+   * @param handle The address of the underlying
+   *   native object.
+   *
+   * @return Java typed access to the data.
+   */
+  protected abstract T data0(long handle);
+
+  /**
+   * Return the length (in bytes) of the data.
+   *
+   * @return The length in bytes.
+   */
+  public int size() {
+    assert (isInitialized());
+    return size0(nativeHandle_);
+  }
+
+  /**
+   * Return true if the length of the
+   * data is zero.
+   *
+   * @return true if there is no data, false otherwise.
+   */
+  public boolean empty() {
+    assert (isInitialized());
+    return empty0(nativeHandle_);
+  }
+
+  /**
+   * Creates a string representation of the data
+   *
+   * @param hex When true, the representation
+   *   will be encoded in hexadecimal.
+   *
+   * @return The string representation of the data.
+   */
+  public String toString(final boolean hex) {
+    assert (isInitialized());
+    return toString0(nativeHandle_, hex);
+  }
+
+  @Override
+  public String toString() {
+    return toString(false);
+  }
+
+  /**
+   * Three-way key comparison
+   *
+   *  @param other A slice to compare against
+   *
+   *  @return Should return either:
+   *    1) < 0 if this < other
+   *    2) == 0 if this == other
+   *    3) > 0 if this > other
+   */
+  public int compare(final AbstractSlice other) {
+    assert (other != null);
+    assert (isInitialized());
+    return compare0(nativeHandle_, other.nativeHandle_);
+  }
+
+  /**
+   * If other is a slice object, then
+   * we defer to {@link #compare(AbstractSlice) compare}
+   * to check equality, otherwise we return false.
+   *
+   * @param other Object to test for equality
+   *
+   * @return true when {@code this.compare(other) == 0},
+   *   false otherwise.
+   */
+  @Override
+  public boolean equals(final Object other) {
+    if (other != null && other instanceof AbstractSlice) {
+      return compare((AbstractSlice)other) == 0;
+    } else {
+      return false;
+    }
+  }
+
+  /**
+   * Determines whether this slice starts with
+   * another slice
+   *
+   * @param prefix Another slice which may of may not
+   *   be a prefix of this slice.
+   *
+   * @return true when this slice starts with the
+   *   {@code prefix} slice
+   */
+  public boolean startsWith(final AbstractSlice prefix) {
+    if (prefix != null) {
+      assert (isInitialized());
+      return startsWith0(nativeHandle_, prefix.nativeHandle_);
+    } else {
+      return false;
+    }
+  }
+
+  /**
+   * Deletes underlying C++ slice pointer.
+   * <p/>
+   * Note that this function should be called only after all
+   * RocksDB instances referencing the slice are closed.
+   * Otherwise an undefined behavior will occur.
+   */
+  @Override
+  protected void disposeInternal() {
+    assert(isInitialized());
+    disposeInternal(nativeHandle_);
+  }
+
+  protected native void createNewSliceFromString(String str);
+  private native int size0(long handle);
+  private native boolean empty0(long handle);
+  private native String toString0(long handle, boolean hex);
+  private native int compare0(long handle, long otherHandle);
+  private native boolean startsWith0(long handle, long otherHandle);
+  private native void disposeInternal(long handle);
+
+}
diff --git a/java/org/rocksdb/Comparator.java b/java/org/rocksdb/Comparator.java
new file mode 100644
index 000000000..c8e050bca
--- /dev/null
+++ b/java/org/rocksdb/Comparator.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;
+
+/**
+ * Base class for comparators which will receive
+ * byte[] based access via org.rocksdb.Slice in their
+ * compare method implementation.
+ *
+ * byte[] based slices perform better when small keys
+ * are involved. When using larger keys consider
+ * using @see org.rocksdb.DirectComparator
+ */
+public abstract class Comparator extends AbstractComparator<Slice> {
+  public Comparator(final ComparatorOptions copt) {
+    super();
+    createNewComparator0(copt.nativeHandle_);
+  }
+
+  private native void createNewComparator0(final long comparatorOptionsHandle);
+}
diff --git a/java/org/rocksdb/ComparatorOptions.java b/java/org/rocksdb/ComparatorOptions.java
new file mode 100644
index 000000000..f0ba520a3
--- /dev/null
+++ b/java/org/rocksdb/ComparatorOptions.java
@@ -0,0 +1,57 @@
+package org.rocksdb;
+
+/**
+ * This class controls the behaviour
+ * of Java implementations of
+ * AbstractComparator
+ *
+ * Note that dispose() must be called before a ComparatorOptions
+ * instance becomes out-of-scope to release the allocated memory in C++.
+ */
+public class ComparatorOptions extends RocksObject {
+  public ComparatorOptions() {
+    super();
+    newComparatorOptions();
+  }
+
+  /**
+   * Use adaptive mutex, which spins in the user space before resorting
+   * to kernel. This could reduce context switch when the mutex is not
+   * heavily contended. However, if the mutex is hot, we could end up
+   * wasting spin time.
+   * Default: false
+   *
+   * @return true if adaptive mutex is used.
+   */
+  public boolean useAdaptiveMutex() {
+    assert(isInitialized());
+    return useAdaptiveMutex(nativeHandle_);
+  }
+
+  /**
+   * Use adaptive mutex, which spins in the user space before resorting
+   * to kernel. This could reduce context switch when the mutex is not
+   * heavily contended. However, if the mutex is hot, we could end up
+   * wasting spin time.
+   * Default: false
+   *
+   * @param useAdaptiveMutex true if adaptive mutex is used.
+   * @return the reference to the current comparator options.
+   */
+  public ComparatorOptions setUseAdaptiveMutex(final boolean useAdaptiveMutex) {
+    assert (isInitialized());
+    setUseAdaptiveMutex(nativeHandle_, useAdaptiveMutex);
+    return this;
+  }
+
+  @Override protected void disposeInternal() {
+    assert(isInitialized());
+    disposeInternal(nativeHandle_);
+  }
+
+  private native void newComparatorOptions();
+  private native boolean useAdaptiveMutex(final long handle);
+  private native void setUseAdaptiveMutex(final long handle,
+      final boolean useAdaptiveMutex);
+  private native void disposeInternal(long handle);
+}
diff --git a/java/org/rocksdb/DirectComparator.java b/java/org/rocksdb/DirectComparator.java
new file mode 100644
index 000000000..47f4d7256
--- /dev/null
+++ b/java/org/rocksdb/DirectComparator.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;
+
+/**
+ * Base class for comparators which will receive
+ * ByteBuffer based access via org.rocksdb.DirectSlice
+ * in their compare method implementation.
+ *
+ * ByteBuffer based slices perform better when large keys
+ * are involved. When using smaller keys consider
+ * using @see org.rocksdb.Comparator
+ */
+public abstract class DirectComparator extends AbstractComparator<DirectSlice> {
+  public DirectComparator(final ComparatorOptions copt) {
+    super();
+    createNewDirectComparator0(copt.nativeHandle_);
+  }
+
+  private native void createNewDirectComparator0(final long comparatorOptionsHandle);
+}
diff --git a/java/org/rocksdb/DirectSlice.java b/java/org/rocksdb/DirectSlice.java
new file mode 100644
index 000000000..847bbd9c1
--- /dev/null
+++ b/java/org/rocksdb/DirectSlice.java
@@ -0,0 +1,113 @@
+// 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;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Base class for slices which will receive direct
+ * ByteBuffer based access to the underlying data.
+ *
+ * ByteBuffer backed slices typically perform better with
+ * larger keys and values. When using smaller keys and
+ * values consider using @see org.rocksdb.Slice
+ */
+public class DirectSlice extends AbstractSlice<ByteBuffer> {
+  /**
+   * Called from JNI to construct a new Java DirectSlice
+   * without an underlying C++ object set
+   * at creation time.
+   *
+   * Note: You should be aware that
+   * {@see org.rocksdb.RocksObject#disOwnNativeHandle()} is intentionally
+   * called from the default DirectSlice constructor, and that it is marked as
+   * private. This is so that developers cannot construct their own default
+   * DirectSlice objects (at present). As developers cannot construct their own
+   * DirectSlice objects through this, they are not creating underlying C++
+   * DirectSlice objects, and so there is nothing to free (dispose) from Java.
+   */
+  private DirectSlice() {
+    super();
+    disOwnNativeHandle();
+  }
+
+  /**
+   * Constructs a slice
+   * where the data is taken from
+   * a String.
+   *
+   * @param str The string
+   */
+  public DirectSlice(final String str) {
+    super();
+    createNewSliceFromString(str);
+  }
+
+  /**
+   * Constructs a slice where the data is
+   * read from the provided
+   * ByteBuffer up to a certain length
+   *
+   * @param data The buffer containing the data
+   * @param length The length of the data to use for the slice
+   */
+  public DirectSlice(final ByteBuffer data, final int length) {
+    super();
+    createNewDirectSlice0(data, length);
+  }
+
+  /**
+   * Constructs a slice where the data is
+   * read from the provided
+   * ByteBuffer
+   *
+   * @param data The bugger containing the data
+   */
+  public DirectSlice(final ByteBuffer data) {
+    super();
+    createNewDirectSlice1(data);
+  }
+
+  /**
+   * Retrieves the byte at a specific offset
+   * from the underlying data
+   *
+   * @param offset The (zero-based) offset of the byte to retrieve
+   *
+   * @return the requested byte
+   */
+  public byte get(int offset) {
+    assert (isInitialized());
+    return get0(nativeHandle_, offset);
+  }
+
+  /**
+   * Clears the backing slice
+   */
+  public void clear() {
+    assert (isInitialized());
+    clear0(nativeHandle_);
+  }
+
+  /**
+   * Drops the specified {@code n}
+   * number of bytes from the start
+   * of the backing slice
+   *
+   * @param n The number of bytes to drop
+   */
+  public void removePrefix(final int n) {
+    assert (isInitialized());
+    removePrefix0(nativeHandle_, n);
+  }
+
+  private native void createNewDirectSlice0(ByteBuffer data, int length);
+  private native void createNewDirectSlice1(ByteBuffer data);
+  @Override protected final native ByteBuffer data0(long handle);
+  private native byte get0(long handle, int offset);
+  private native void clear0(long handle);
+  private native void removePrefix0(long handle, int length);
+}
diff --git a/java/org/rocksdb/Options.java b/java/org/rocksdb/Options.java
index 741404e40..b99d0c7ea 100644
--- a/java/org/rocksdb/Options.java
+++ b/java/org/rocksdb/Options.java
@@ -193,6 +193,27 @@ public class Options extends RocksObject {
     return maxWriteBufferNumber(nativeHandle_);
   }
 
+  /**
+   * Use the specified comparator for key ordering.
+   *
+   * Comparator should not be disposed before options instances using this comparator is
+   * disposed. If dispose() function is not called, then comparator object will be
+   * GC'd automatically.
+   *
+   * Comparator instance can be re-used in multiple options instances.
+   *
+   * @param comparator java instance.
+   * @return the instance of the current Options.
+   * @see RocksDB.open()
+   */
+  public Options setComparator(AbstractComparator comparator) {
+      assert (isInitialized());
+      setComparatorHandle(nativeHandle_, comparator.nativeHandle_);
+      comparator_ = comparator;
+      return this;
+  }
+  private native void setComparatorHandle(long optHandle, long comparatorHandle);
+
   /**
    * If true, an error will be thrown during RocksDB.open() if the
    * database already exists.
@@ -2282,6 +2303,7 @@ public class Options extends RocksObject {
 
   long cacheSize_;
   int numShardBits_;
+  AbstractComparator comparator_;
   RocksEnv env_;
   MemTableConfig memTableConfig_;
   TableFormatConfig tableFormatConfig_;
diff --git a/java/org/rocksdb/Slice.java b/java/org/rocksdb/Slice.java
new file mode 100644
index 000000000..4449cb7b8
--- /dev/null
+++ b/java/org/rocksdb/Slice.java
@@ -0,0 +1,84 @@
+// 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;
+
+/**
+ * Base class for slices which will receive
+ * byte[] based access to the underlying data.
+ *
+ * byte[] backed slices typically perform better with
+ * small keys and values. When using larger keys and
+ * values consider using @see org.rocksdb.DirectSlice
+ */
+public class Slice extends AbstractSlice<byte[]> {
+  /**
+   * Called from JNI to construct a new Java Slice
+   * without an underlying C++ object set
+   * at creation time.
+   *
+   * Note: You should be aware that
+   * {@see org.rocksdb.RocksObject#disOwnNativeHandle()} is intentionally
+   * called from the default Slice constructor, and that it is marked as
+   * private. This is so that developers cannot construct their own default
+   * Slice objects (at present). As developers cannot construct their own
+   * Slice objects through this, they are not creating underlying C++ Slice
+   * objects, and so there is nothing to free (dispose) from Java.
+   */
+  private Slice() {
+    super();
+    disOwnNativeHandle();
+  }
+
+  /**
+   * Constructs a slice
+   * where the data is taken from
+   * a String.
+   */
+  public Slice(final String str) {
+    super();
+    createNewSliceFromString(str);
+  }
+
+  /**
+   * Constructs a slice
+   * where the data is a copy of
+   * the byte array from a specific offset.
+   */
+  public Slice(final byte[] data, final int offset) {
+    super();
+    createNewSlice0(data, offset);
+  }
+
+  /**
+   * Constructs a slice
+   * where the data is a copy of
+   * the byte array.
+   */
+  public Slice(final byte[] data) {
+    super();
+    createNewSlice1(data);
+  }
+
+  /**
+   * Deletes underlying C++ slice pointer
+   * and any buffered data.
+   *
+   * <p/>
+   * Note that this function should be called only after all
+   * RocksDB instances referencing the slice are closed.
+   * Otherwise an undefined behavior will occur.
+   */
+  @Override
+  protected void disposeInternal() {
+    super.disposeInternal();
+    disposeInternalBuf(nativeHandle_);
+  }
+
+  @Override protected final native byte[] data0(long handle);
+  private native void createNewSlice0(byte[] data, int length);
+  private native void createNewSlice1(byte[] data);
+  private native void disposeInternalBuf(long handle);
+}
diff --git a/java/org/rocksdb/test/AbstractComparatorTest.java b/java/org/rocksdb/test/AbstractComparatorTest.java
new file mode 100644
index 000000000..dfdb3cad9
--- /dev/null
+++ b/java/org/rocksdb/test/AbstractComparatorTest.java
@@ -0,0 +1,166 @@
+// 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 org.rocksdb.*;
+
+import java.io.IOException;
+import java.nio.file.*;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Random;
+
+import static org.rocksdb.test.Types.byteToInt;
+import static org.rocksdb.test.Types.intToByte;
+
+/**
+ * Abstract tests for both Comparator and DirectComparator
+ */
+public abstract class AbstractComparatorTest {
+
+  /**
+   * Get a comparator which will expect Integer keys
+   * and determine an ascending order
+   *
+   * @return An integer ascending order key comparator
+   */
+  public abstract AbstractComparator getAscendingIntKeyComparator();
+
+  /**
+   * Test which stores random keys into the database
+   * using an @see getAscendingIntKeyComparator
+   * it then checks that these keys are read back in
+   * ascending order
+   *
+   * @param db_path A path where we can store database
+   *                files temporarily
+   */
+  public void testRoundtrip(final Path db_path) throws IOException {
+
+    Options opt = null;
+    RocksDB db = null;
+
+    try {
+      opt = new Options();
+      opt.setCreateIfMissing(true);
+      opt.setComparator(getAscendingIntKeyComparator());
+
+      // store 10,000 random integer keys
+      final int ITERATIONS = 10000;
+
+      db = RocksDB.open(opt, db_path.toString());
+      final Random random = new Random();
+      for (int i = 0; i < ITERATIONS; i++) {
+        final byte key[] = intToByte(random.nextInt());
+        if (i > 0 && db.get(key) != null) { // does key already exist (avoid duplicates)
+          i--; // generate a different key
+        } else {
+          db.put(key, "value".getBytes());
+        }
+      }
+      db.close();
+
+
+      // re-open db and read from start to end
+      // integer keys should be in ascending
+      // order as defined by SimpleIntComparator
+      db = RocksDB.open(opt, db_path.toString());
+      final RocksIterator it = db.newIterator();
+      it.seekToFirst();
+      int lastKey = Integer.MIN_VALUE;
+      int count = 0;
+      for (it.seekToFirst(); it.isValid(); it.next()) {
+        final int thisKey = byteToInt(it.key());
+        assert(thisKey > lastKey);
+        lastKey = thisKey;
+        count++;
+      }
+      db.close();
+
+      assert(count == ITERATIONS);
+
+    } catch (final RocksDBException e) {
+      System.err.format("[ERROR]: %s%n", e);
+      e.printStackTrace();
+    } finally {
+      if (db != null) {
+        db.close();
+      }
+
+      if (opt != null) {
+        opt.dispose();
+      }
+
+      removeDb(db_path); // cleanup after ourselves!
+    }
+  }
+
+  /**
+   * Compares integer keys
+   * so that they are in ascending order
+   *
+   * @param a 4-bytes representing an integer key
+   * @param b 4-bytes representing an integer key
+   *
+   * @return negative if a < b, 0 if a == b, positive otherwise
+   */
+  protected final int compareIntKeys(final byte[] a, final byte[] b) {
+
+    final int iA = byteToInt(a);
+    final int iB = byteToInt(b);
+
+    // protect against int key calculation overflow
+    final double diff = (double)iA - iB;
+    final int result;
+    if (diff < Integer.MIN_VALUE) {
+      result = Integer.MIN_VALUE;
+    } else if(diff > Integer.MAX_VALUE) {
+      result = Integer.MAX_VALUE;
+    } else {
+      result = (int)diff;
+    }
+
+    return result;
+  }
+
+  /**
+   * Utility method for deleting database files
+   *
+   * @param db_path The path to the database to remove
+   *                from the filesystem
+   */
+  private static void removeDb(final Path db_path) throws IOException {
+    Files.walkFileTree(db_path, new SimpleFileVisitor<Path>() {
+      @Override
+      public FileVisitResult visitFile(final Path file, final BasicFileAttributes attrs)
+          throws IOException {
+        Files.delete(file);
+        return FileVisitResult.CONTINUE;
+      }
+
+      @Override
+      public FileVisitResult visitFileFailed(final Path file, IOException exc)
+          throws IOException {
+        // try to delete the file anyway, even if its attributes
+        // could not be read, since delete-only access is
+        // theoretically possible
+        Files.delete(file);
+        return FileVisitResult.CONTINUE;
+      }
+
+      @Override
+      public FileVisitResult postVisitDirectory(final Path dir, IOException exc)
+          throws IOException {
+        if (exc == null) {
+          Files.delete(dir);
+          return FileVisitResult.CONTINUE;
+        } else {
+          // directory iteration failed; propagate exception
+          throw exc;
+        }
+      }
+    });
+  }
+}
diff --git a/java/org/rocksdb/test/ComparatorOptionsTest.java b/java/org/rocksdb/test/ComparatorOptionsTest.java
new file mode 100644
index 000000000..e25209392
--- /dev/null
+++ b/java/org/rocksdb/test/ComparatorOptionsTest.java
@@ -0,0 +1,34 @@
+// 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 org.rocksdb.ComparatorOptions;
+import org.rocksdb.RocksDB;
+
+import java.util.Random;
+
+public class ComparatorOptionsTest {
+
+  static {
+    RocksDB.loadLibrary();
+  }
+
+  public static void main(String[] args) {
+    final ComparatorOptions copt = new ComparatorOptions();
+    Random rand = new Random();
+
+    { // UseAdaptiveMutex test
+      copt.setUseAdaptiveMutex(true);
+      assert(copt.useAdaptiveMutex() == true);
+
+      copt.setUseAdaptiveMutex(false);
+      assert(copt.useAdaptiveMutex() == false);
+    }
+
+    copt.dispose();
+    System.out.println("Passed ComparatorOptionsTest");
+  }
+}
diff --git a/java/org/rocksdb/test/ComparatorTest.java b/java/org/rocksdb/test/ComparatorTest.java
new file mode 100644
index 000000000..34d7c78df
--- /dev/null
+++ b/java/org/rocksdb/test/ComparatorTest.java
@@ -0,0 +1,45 @@
+// 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 org.rocksdb.*;
+
+import java.io.IOException;
+import java.nio.file.FileSystems;
+
+public class ComparatorTest {
+  private static final String db_path = "/tmp/comparator_db";
+
+  static {
+    RocksDB.loadLibrary();
+  }
+
+  public static void main(String[] args) throws IOException {
+
+    final AbstractComparatorTest comparatorTest = new AbstractComparatorTest() {
+      @Override
+      public AbstractComparator getAscendingIntKeyComparator() {
+        return new Comparator(new ComparatorOptions()) {
+
+          @Override
+          public String name() {
+            return "test.AscendingIntKeyComparator";
+          }
+
+          @Override
+          public int compare(final Slice a, final Slice b) {
+            return compareIntKeys(a.data(), b.data());
+          }
+        };
+      }
+    };
+
+    // test the round-tripability of keys written and read with the Comparator
+    comparatorTest.testRoundtrip(FileSystems.getDefault().getPath(db_path));
+
+    System.out.println("Passed ComparatorTest");
+  }
+}
diff --git a/java/org/rocksdb/test/DirectComparatorTest.java b/java/org/rocksdb/test/DirectComparatorTest.java
new file mode 100644
index 000000000..9df06eb73
--- /dev/null
+++ b/java/org/rocksdb/test/DirectComparatorTest.java
@@ -0,0 +1,48 @@
+// 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 org.rocksdb.*;
+
+import java.io.IOException;
+import java.nio.file.FileSystems;
+
+public class DirectComparatorTest {
+  private static final String db_path = "/tmp/direct_comparator_db";
+
+  static {
+    RocksDB.loadLibrary();
+  }
+
+  public static void main(String[] args) throws IOException {
+
+    final AbstractComparatorTest comparatorTest = new AbstractComparatorTest() {
+      @Override
+      public AbstractComparator getAscendingIntKeyComparator() {
+        return new DirectComparator(new ComparatorOptions()) {
+
+          @Override
+          public String name() {
+            return "test.AscendingIntKeyDirectComparator";
+          }
+
+          @Override
+          public int compare(final DirectSlice a, final DirectSlice b) {
+            final byte ax[] = new byte[4], bx[] = new byte[4];
+            a.data().get(ax);
+            b.data().get(bx);
+            return compareIntKeys(ax, bx);
+          }
+        };
+      }
+    };
+
+    // test the round-tripability of keys written and read with the DirectComparator
+    comparatorTest.testRoundtrip(FileSystems.getDefault().getPath(db_path));
+
+    System.out.println("Passed DirectComparatorTest");
+  }
+}
diff --git a/java/org/rocksdb/test/Types.java b/java/org/rocksdb/test/Types.java
new file mode 100644
index 000000000..22fcd3537
--- /dev/null
+++ b/java/org/rocksdb/test/Types.java
@@ -0,0 +1,43 @@
+// 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;
+
+/**
+ * Simple type conversion methods
+ * for use in tests
+ */
+public class Types {
+
+  /**
+   * Convert first 4 bytes of a byte array to an int
+   *
+   * @param data The byte array
+   *
+   * @return An integer
+   */
+  public static int byteToInt(final byte data[]) {
+    return (data[0] & 0xff) |
+        ((data[1] & 0xff) << 8) |
+        ((data[2] & 0xff) << 16) |
+        ((data[3] & 0xff) << 24);
+  }
+
+  /**
+   * Convert an int to 4 bytes
+   *
+   * @param v The int
+   *
+   * @return A byte array containing 4 bytes
+   */
+  public static byte[] intToByte(final int v) {
+    return new byte[] {
+        (byte)((v >>> 0) & 0xff),
+        (byte)((v >>> 8) & 0xff),
+        (byte)((v >>> 16) & 0xff),
+        (byte)((v >>> 24) & 0xff)
+    };
+  }
+}
diff --git a/java/rocksjni/comparator.cc b/java/rocksjni/comparator.cc
new file mode 100644
index 000000000..196376235
--- /dev/null
+++ b/java/rocksjni/comparator.cc
@@ -0,0 +1,66 @@
+// 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.
+//
+// This file implements the "bridge" between Java and C++ for
+// rocksdb::Comparator.
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <jni.h>
+#include <string>
+#include <functional>
+
+#include "include/org_rocksdb_AbstractComparator.h"
+#include "include/org_rocksdb_Comparator.h"
+#include "include/org_rocksdb_DirectComparator.h"
+#include "rocksjni/comparatorjnicallback.h"
+#include "rocksjni/portal.h"
+
+// <editor-fold desc="org.rocksdb.AbstractComparator>
+
+/*
+ * Class:     org_rocksdb_AbstractComparator
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_AbstractComparator_disposeInternal(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  delete reinterpret_cast<rocksdb::BaseComparatorJniCallback*>(handle);
+}
+// </editor-fold>
+
+// <editor-fold desc="org.rocksdb.Comparator>
+
+/*
+ * Class:     org_rocksdb_Comparator
+ * Method:    createNewComparator0
+ * Signature: ()V
+ */
+void Java_org_rocksdb_Comparator_createNewComparator0(
+    JNIEnv* env, jobject jobj, jlong copt_handle) {
+  const rocksdb::ComparatorJniCallbackOptions* copt =
+    reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(copt_handle);
+  const rocksdb::ComparatorJniCallback* c =
+    new rocksdb::ComparatorJniCallback(env, jobj, copt);
+  rocksdb::AbstractComparatorJni::setHandle(env, jobj, c);
+}
+// </editor-fold>
+
+// <editor-fold desc="org.rocksdb.DirectComparator>
+
+/*
+ * Class:     org_rocksdb_DirectComparator
+ * Method:    createNewDirectComparator0
+ * Signature: ()V
+ */
+void Java_org_rocksdb_DirectComparator_createNewDirectComparator0(
+    JNIEnv* env, jobject jobj, jlong copt_handle) {
+  const rocksdb::ComparatorJniCallbackOptions* copt =
+    reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(copt_handle);
+  const rocksdb::DirectComparatorJniCallback* c =
+    new rocksdb::DirectComparatorJniCallback(env, jobj, copt);
+  rocksdb::AbstractComparatorJni::setHandle(env, jobj, c);
+}
+// </editor-fold>
diff --git a/java/rocksjni/comparatorjnicallback.cc b/java/rocksjni/comparatorjnicallback.cc
new file mode 100644
index 000000000..1be363541
--- /dev/null
+++ b/java/rocksjni/comparatorjnicallback.cc
@@ -0,0 +1,176 @@
+// 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.
+//
+// This file implements the callback "bridge" between Java and C++ for
+// rocksdb::Comparator.
+
+#include "rocksjni/comparatorjnicallback.h"
+#include "rocksjni/portal.h"
+
+namespace rocksdb {
+BaseComparatorJniCallback::BaseComparatorJniCallback(
+    JNIEnv* env, jobject jComparator,
+    const ComparatorJniCallbackOptions* copt)
+    : mtx_compare(new port::Mutex(copt->use_adaptive_mutex)),
+    mtx_findShortestSeparator(new port::Mutex(copt->use_adaptive_mutex)) {
+
+  // Note: Comparator methods may be accessed by multiple threads,
+  // so we ref the jvm not the env
+  const jint rs = env->GetJavaVM(&m_jvm);
+  assert(rs == JNI_OK);
+
+  // Note: we want to access the Java Comparator instance
+  // across multiple method calls, so we create a global ref
+  m_jComparator = env->NewGlobalRef(jComparator);
+
+  // Note: The name of a Comparator will not change during it's lifetime,
+  // so we cache it in a global var
+  jmethodID jNameMethodId = AbstractComparatorJni::getNameMethodId(env);
+  jstring jsName = (jstring)env->CallObjectMethod(m_jComparator, jNameMethodId);
+  m_name = JniUtil::copyString(env, jsName);  // also releases jsName
+
+  m_jCompareMethodId = AbstractComparatorJni::getCompareMethodId(env);
+  m_jFindShortestSeparatorMethodId =
+    AbstractComparatorJni::getFindShortestSeparatorMethodId(env);
+  m_jFindShortSuccessorMethodId =
+    AbstractComparatorJni::getFindShortSuccessorMethodId(env);
+}
+
+/**
+ * Attach/Get a JNIEnv for the current native thread
+ */
+JNIEnv* BaseComparatorJniCallback::getJniEnv() const {
+  JNIEnv *env;
+  jint rs = m_jvm->AttachCurrentThread(reinterpret_cast<void **>(&env), NULL);
+  assert(rs == JNI_OK);
+  return env;
+}
+
+const char* BaseComparatorJniCallback::Name() const {
+  return m_name.c_str();
+}
+
+int BaseComparatorJniCallback::Compare(const Slice& a, const Slice& b) const {
+  JNIEnv* m_env = getJniEnv();
+
+  // TODO(adamretter): slice objects can potentially be cached using thread
+  // local variables to avoid locking. Could make this configurable depending on
+  // performance.
+  mtx_compare->Lock();
+
+  AbstractSliceJni::setHandle(m_env, m_jSliceA, &a);
+  AbstractSliceJni::setHandle(m_env, m_jSliceB, &b);
+  jint result =
+    m_env->CallIntMethod(m_jComparator, m_jCompareMethodId, m_jSliceA,
+      m_jSliceB);
+
+  mtx_compare->Unlock();
+
+  m_jvm->DetachCurrentThread();
+
+  return result;
+}
+
+void BaseComparatorJniCallback::FindShortestSeparator(
+  std::string* start, const Slice& limit) const {
+  if (start == nullptr) {
+    return;
+  }
+
+  JNIEnv* m_env = getJniEnv();
+
+  const char* startUtf = start->c_str();
+  jstring jsStart = m_env->NewStringUTF(startUtf);
+
+  // TODO(adamretter): slice object can potentially be cached using thread local
+  // variable to avoid locking. Could make this configurable depending on
+  // performance.
+  mtx_findShortestSeparator->Lock();
+
+  AbstractSliceJni::setHandle(m_env, m_jSliceLimit, &limit);
+  jstring jsResultStart =
+    (jstring)m_env->CallObjectMethod(m_jComparator,
+      m_jFindShortestSeparatorMethodId, jsStart, m_jSliceLimit);
+
+  mtx_findShortestSeparator->Unlock();
+
+  m_env->DeleteLocalRef(jsStart);
+
+  if (jsResultStart != nullptr) {
+    // update start with result
+    *start =
+      JniUtil::copyString(m_env, jsResultStart);  // also releases jsResultStart
+  }
+
+  m_jvm->DetachCurrentThread();
+}
+
+void BaseComparatorJniCallback::FindShortSuccessor(std::string* key) const {
+  if (key == nullptr) {
+    return;
+  }
+
+  JNIEnv* m_env = getJniEnv();
+
+  const char* keyUtf = key->c_str();
+  jstring jsKey = m_env->NewStringUTF(keyUtf);
+
+  jstring jsResultKey =
+    (jstring)m_env->CallObjectMethod(m_jComparator,
+      m_jFindShortSuccessorMethodId, jsKey);
+
+  m_env->DeleteLocalRef(jsKey);
+
+  if (jsResultKey != nullptr) {
+    // updates key with result, also releases jsResultKey.
+    *key = JniUtil::copyString(m_env, jsResultKey);
+  }
+
+  m_jvm->DetachCurrentThread();
+}
+
+BaseComparatorJniCallback::~BaseComparatorJniCallback() {
+  JNIEnv* m_env = getJniEnv();
+
+  m_env->DeleteGlobalRef(m_jComparator);
+
+  // Note: do not need to explicitly detach, as this function is effectively
+  // called from the Java class's disposeInternal method, and so already
+  // has an attached thread, getJniEnv above is just a no-op Attach to get
+  // the env jvm->DetachCurrentThread();
+}
+
+ComparatorJniCallback::ComparatorJniCallback(
+    JNIEnv* env, jobject jComparator,
+    const ComparatorJniCallbackOptions* copt) :
+    BaseComparatorJniCallback(env, jComparator, copt) {
+  m_jSliceA = env->NewGlobalRef(SliceJni::construct0(env));
+  m_jSliceB = env->NewGlobalRef(SliceJni::construct0(env));
+  m_jSliceLimit = env->NewGlobalRef(SliceJni::construct0(env));
+}
+
+ComparatorJniCallback::~ComparatorJniCallback() {
+  JNIEnv* m_env = getJniEnv();
+  m_env->DeleteGlobalRef(m_jSliceA);
+  m_env->DeleteGlobalRef(m_jSliceB);
+  m_env->DeleteGlobalRef(m_jSliceLimit);
+}
+
+DirectComparatorJniCallback::DirectComparatorJniCallback(
+    JNIEnv* env, jobject jComparator,
+    const ComparatorJniCallbackOptions* copt) :
+    BaseComparatorJniCallback(env, jComparator, copt) {
+  m_jSliceA = env->NewGlobalRef(DirectSliceJni::construct0(env));
+  m_jSliceB = env->NewGlobalRef(DirectSliceJni::construct0(env));
+  m_jSliceLimit = env->NewGlobalRef(DirectSliceJni::construct0(env));
+}
+
+DirectComparatorJniCallback::~DirectComparatorJniCallback() {
+  JNIEnv* m_env = getJniEnv();
+  m_env->DeleteGlobalRef(m_jSliceA);
+  m_env->DeleteGlobalRef(m_jSliceB);
+  m_env->DeleteGlobalRef(m_jSliceLimit);
+}
+}  // namespace rocksdb
diff --git a/java/rocksjni/comparatorjnicallback.h b/java/rocksjni/comparatorjnicallback.h
new file mode 100644
index 000000000..65b986ca4
--- /dev/null
+++ b/java/rocksjni/comparatorjnicallback.h
@@ -0,0 +1,95 @@
+// 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.
+//
+// This file implements the callback "bridge" between Java and C++ for
+// rocksdb::Comparator and rocksdb::DirectComparator.
+
+#ifndef JAVA_ROCKSJNI_COMPARATORJNICALLBACK_H_
+#define JAVA_ROCKSJNI_COMPARATORJNICALLBACK_H_
+
+#include <jni.h>
+#include <string>
+#include "rocksdb/comparator.h"
+#include "rocksdb/slice.h"
+#include "port/port.h"
+
+namespace rocksdb {
+
+struct ComparatorJniCallbackOptions {
+  // Use adaptive mutex, which spins in the user space before resorting
+  // to kernel. This could reduce context switch when the mutex is not
+  // heavily contended. However, if the mutex is hot, we could end up
+  // wasting spin time.
+  // Default: false
+  bool use_adaptive_mutex;
+
+  ComparatorJniCallbackOptions() : use_adaptive_mutex(false) {
+  }
+};
+
+/**
+ * This class acts as a bridge between C++
+ * and Java. The methods in this class will be
+ * called back from the RocksDB storage engine (C++)
+ * we then callback to the appropriate Java method
+ * this enables Comparators to be implemented in Java.
+ *
+ * The design of this Comparator caches the Java Slice
+ * objects that are used in the compare and findShortestSeparator
+ * method callbacks. Instead of creating new objects for each callback
+ * of those functions, by reuse via setHandle we are a lot
+ * faster; Unfortunately this means that we have to
+ * introduce independent locking in regions of each of those methods
+ * via the mutexs mtx_compare and mtx_findShortestSeparator respectively
+ */
+class BaseComparatorJniCallback : public Comparator {
+ public:
+    BaseComparatorJniCallback(
+      JNIEnv* env, jobject jComparator,
+      const ComparatorJniCallbackOptions* copt);
+    virtual ~BaseComparatorJniCallback();
+    virtual const char* Name() const;
+    virtual int Compare(const Slice& a, const Slice& b) const;
+    virtual void FindShortestSeparator(
+      std::string* start, const Slice& limit) const;
+    virtual void FindShortSuccessor(std::string* key) const;
+
+ private:
+    // used for synchronisation in compare method
+    port::Mutex* mtx_compare;
+    // used for synchronisation in findShortestSeparator method
+    port::Mutex* mtx_findShortestSeparator;
+    JavaVM* m_jvm;
+    jobject m_jComparator;
+    std::string m_name;
+    jmethodID m_jCompareMethodId;
+    jmethodID m_jFindShortestSeparatorMethodId;
+    jmethodID m_jFindShortSuccessorMethodId;
+
+ protected:
+    JNIEnv* getJniEnv() const;
+    jobject m_jSliceA;
+    jobject m_jSliceB;
+    jobject m_jSliceLimit;
+};
+
+class ComparatorJniCallback : public BaseComparatorJniCallback {
+ public:
+      ComparatorJniCallback(
+        JNIEnv* env, jobject jComparator,
+        const ComparatorJniCallbackOptions* copt);
+      ~ComparatorJniCallback();
+};
+
+class DirectComparatorJniCallback : public BaseComparatorJniCallback {
+ public:
+      DirectComparatorJniCallback(
+        JNIEnv* env, jobject jComparator,
+        const ComparatorJniCallbackOptions* copt);
+      ~DirectComparatorJniCallback();
+};
+}  // namespace rocksdb
+
+#endif  // JAVA_ROCKSJNI_COMPARATORJNICALLBACK_H_
diff --git a/java/rocksjni/options.cc b/java/rocksjni/options.cc
index ef104d92b..ceb4ce031 100644
--- a/java/rocksjni/options.cc
+++ b/java/rocksjni/options.cc
@@ -14,6 +14,9 @@
 #include "include/org_rocksdb_Options.h"
 #include "include/org_rocksdb_WriteOptions.h"
 #include "include/org_rocksdb_ReadOptions.h"
+#include "include/org_rocksdb_ComparatorOptions.h"
+
+#include "rocksjni/comparatorjnicallback.h"
 #include "rocksjni/portal.h"
 #include "rocksdb/db.h"
 #include "rocksdb/options.h"
@@ -164,6 +167,17 @@ jlong Java_org_rocksdb_Options_statisticsPtr(
   return reinterpret_cast<jlong>(st);
 }
 
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    setComparatorHandle
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_Options_setComparatorHandle(
+    JNIEnv* env, jobject jobj, jlong jopt_handle, jlong jcomparator_handle) {
+  reinterpret_cast<rocksdb::Options*>(jopt_handle)->comparator =
+      reinterpret_cast<rocksdb::Comparator*>(jcomparator_handle);
+}
+
 /*
  * Class:     org_rocksdb_Options
  * Method:    maxWriteBufferNumber
@@ -1759,3 +1773,49 @@ void Java_org_rocksdb_ReadOptions_setTailing(
   reinterpret_cast<rocksdb::ReadOptions*>(jhandle)->tailing =
       static_cast<bool>(jtailing);
 }
+
+/////////////////////////////////////////////////////////////////////
+// rocksdb::ComparatorOptions
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    newComparatorOptions
+ * Signature: ()V
+ */
+void Java_org_rocksdb_ComparatorOptions_newComparatorOptions(
+    JNIEnv* env, jobject jobj) {
+  auto comparator_opt = new rocksdb::ComparatorJniCallbackOptions();
+  rocksdb::ComparatorOptionsJni::setHandle(env, jobj, comparator_opt);
+}
+
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    useAdaptiveMutex
+ * Signature: (J)Z
+ */
+jboolean Java_org_rocksdb_ComparatorOptions_useAdaptiveMutex(
+    JNIEnv * env, jobject jobj, jlong jhandle) {
+  return reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle)
+    ->use_adaptive_mutex;
+}
+
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    setUseAdaptiveMutex
+ * Signature: (JZ)V
+ */
+void Java_org_rocksdb_ComparatorOptions_setUseAdaptiveMutex(
+    JNIEnv * env, jobject jobj, jlong jhandle, jboolean juse_adaptive_mutex) {
+  reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle)
+    ->use_adaptive_mutex = static_cast<bool>(juse_adaptive_mutex);
+}
+
+/*
+ * Class:     org_rocksdb_ComparatorOptions
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_ComparatorOptions_disposeInternal(
+    JNIEnv * env, jobject jobj, jlong jhandle) {
+  delete reinterpret_cast<rocksdb::ComparatorJniCallbackOptions*>(jhandle);
+  rocksdb::ComparatorOptionsJni::setHandle(env, jobj, nullptr);
+}
diff --git a/java/rocksjni/portal.h b/java/rocksjni/portal.h
index 8300a6e66..32452ae0b 100644
--- a/java/rocksjni/portal.h
+++ b/java/rocksjni/portal.h
@@ -12,10 +12,13 @@
 
 #include <jni.h>
 #include <limits>
+#include <string>
+
 #include "rocksdb/db.h"
 #include "rocksdb/filter_policy.h"
 #include "rocksdb/status.h"
 #include "rocksdb/utilities/backupable_db.h"
+#include "rocksjni/comparatorjnicallback.h"
 
 namespace rocksdb {
 
@@ -362,6 +365,168 @@ class ColumnFamilyHandleJni {
   }
 };
 
+class ComparatorOptionsJni {
+ public:
+    // Get the java class id of org.rocksdb.ComparatorOptions.
+    static jclass getJClass(JNIEnv* env) {
+      jclass jclazz = env->FindClass("org/rocksdb/ComparatorOptions");
+      assert(jclazz != nullptr);
+      return jclazz;
+    }
+
+    // Get the field id of the member variable of org.rocksdb.ComparatorOptions
+    // that stores the pointer to rocksdb::ComparatorJniCallbackOptions.
+    static jfieldID getHandleFieldID(JNIEnv* env) {
+      static jfieldID fid = env->GetFieldID(
+          getJClass(env), "nativeHandle_", "J");
+      assert(fid != nullptr);
+      return fid;
+    }
+
+    // Pass the ComparatorJniCallbackOptions pointer to the java side.
+    static void setHandle(
+      JNIEnv* env, jobject jobj,
+      const rocksdb::ComparatorJniCallbackOptions* op) {
+      env->SetLongField(
+          jobj, getHandleFieldID(env),
+          reinterpret_cast<jlong>(op));
+    }
+};
+
+class AbstractComparatorJni {
+ public:
+  // Get the java class id of org.rocksdb.Comparator.
+  static jclass getJClass(JNIEnv* env) {
+    jclass jclazz = env->FindClass("org/rocksdb/AbstractComparator");
+    assert(jclazz != nullptr);
+    return jclazz;
+  }
+
+  // Get the field id of the member variable of org.rocksdb.Comparator
+  // that stores the pointer to rocksdb::Comparator.
+  static jfieldID getHandleFieldID(JNIEnv* env) {
+    static jfieldID fid = env->GetFieldID(
+        getJClass(env), "nativeHandle_", "J");
+    assert(fid != nullptr);
+    return fid;
+  }
+
+  // Get the java method `name` of org.rocksdb.Comparator.
+  static jmethodID getNameMethodId(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(
+        getJClass(env), "name", "()Ljava/lang/String;");
+    assert(mid != nullptr);
+    return mid;
+  }
+
+  // Get the java method `compare` of org.rocksdb.Comparator.
+  static jmethodID getCompareMethodId(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(getJClass(env),
+      "compare",
+      "(Lorg/rocksdb/AbstractSlice;Lorg/rocksdb/AbstractSlice;)I");
+    assert(mid != nullptr);
+    return mid;
+  }
+
+  // Get the java method `findShortestSeparator` of org.rocksdb.Comparator.
+  static jmethodID getFindShortestSeparatorMethodId(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(getJClass(env),
+      "findShortestSeparator",
+      "(Ljava/lang/String;Lorg/rocksdb/AbstractSlice;)Ljava/lang/String;");
+    assert(mid != nullptr);
+    return mid;
+  }
+
+  // Get the java method `findShortSuccessor` of org.rocksdb.Comparator.
+  static jmethodID getFindShortSuccessorMethodId(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(getJClass(env),
+      "findShortSuccessor",
+      "(Ljava/lang/String;)Ljava/lang/String;");
+    assert(mid != nullptr);
+    return mid;
+  }
+
+  // Get the pointer to ComparatorJniCallback.
+  static rocksdb::BaseComparatorJniCallback* getHandle(
+    JNIEnv* env, jobject jobj) {
+    return reinterpret_cast<rocksdb::BaseComparatorJniCallback*>(
+        env->GetLongField(jobj, getHandleFieldID(env)));
+  }
+
+  // Pass the ComparatorJniCallback pointer to the java side.
+  static void setHandle(
+    JNIEnv* env, jobject jobj, const rocksdb::BaseComparatorJniCallback* op) {
+    env->SetLongField(
+        jobj, getHandleFieldID(env),
+        reinterpret_cast<jlong>(op));
+  }
+};
+
+class AbstractSliceJni {
+ public:
+  // Get the java class id of org.rocksdb.Slice.
+  static jclass getJClass(JNIEnv* env) {
+    jclass jclazz = env->FindClass("org/rocksdb/AbstractSlice");
+    assert(jclazz != nullptr);
+    return jclazz;
+  }
+
+  // Get the field id of the member variable of org.rocksdb.Slice
+  // that stores the pointer to rocksdb::Slice.
+  static jfieldID getHandleFieldID(JNIEnv* env) {
+    static jfieldID fid = env->GetFieldID(
+        getJClass(env), "nativeHandle_", "J");
+    assert(fid != nullptr);
+    return fid;
+  }
+
+  // Get the pointer to Slice.
+  static rocksdb::Slice* getHandle(JNIEnv* env, jobject jobj) {
+    return reinterpret_cast<rocksdb::Slice*>(
+        env->GetLongField(jobj, getHandleFieldID(env)));
+  }
+
+  // Pass the Slice pointer to the java side.
+  static void setHandle(
+      JNIEnv* env, jobject jobj, const rocksdb::Slice* op) {
+    env->SetLongField(
+        jobj, getHandleFieldID(env),
+        reinterpret_cast<jlong>(op));
+  }
+};
+
+class SliceJni {
+ public:
+  // Get the java class id of org.rocksdb.Slice.
+  static jclass getJClass(JNIEnv* env) {
+    jclass jclazz = env->FindClass("org/rocksdb/Slice");
+    assert(jclazz != nullptr);
+    return jclazz;
+  }
+
+  static jobject construct0(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(getJClass(env), "<init>", "()V");
+    assert(mid != nullptr);
+    return env->NewObject(getJClass(env), mid);
+  }
+};
+
+class DirectSliceJni {
+ public:
+  // Get the java class id of org.rocksdb.DirectSlice.
+  static jclass getJClass(JNIEnv* env) {
+    jclass jclazz = env->FindClass("org/rocksdb/DirectSlice");
+    assert(jclazz != nullptr);
+    return jclazz;
+  }
+
+  static jobject construct0(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(getJClass(env), "<init>", "()V");
+    assert(mid != nullptr);
+    return env->NewObject(getJClass(env), mid);
+  }
+};
+
 class ListJni {
  public:
   // Get the java class id of java.util.List.
@@ -425,5 +590,20 @@ class ListJni {
     return mid;
   }
 };
+
+class JniUtil {
+ public:
+    /**
+     * Copies a jstring to a std::string
+     * and releases the original jstring
+     */
+    static std::string copyString(JNIEnv* env, jstring js) {
+      const char *utf = env->GetStringUTFChars(js, NULL);
+      std::string name(utf);
+      env->ReleaseStringUTFChars(js, utf);
+      return name;
+    }
+};
+
 }  // namespace rocksdb
 #endif  // JAVA_ROCKSJNI_PORTAL_H_
diff --git a/java/rocksjni/slice.cc b/java/rocksjni/slice.cc
new file mode 100644
index 000000000..0d8b92c9c
--- /dev/null
+++ b/java/rocksjni/slice.cc
@@ -0,0 +1,251 @@
+// 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.
+//
+// This file implements the "bridge" between Java and C++ for
+// rocksdb::Slice.
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <jni.h>
+#include <string>
+
+#include "include/org_rocksdb_AbstractSlice.h"
+#include "include/org_rocksdb_Slice.h"
+#include "include/org_rocksdb_DirectSlice.h"
+#include "rocksdb/slice.h"
+#include "rocksjni/portal.h"
+
+// <editor-fold desc="org.rocksdb.AbstractSlice>
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    createNewSliceFromString
+ * Signature: (Ljava/lang/String;)V
+ */
+void Java_org_rocksdb_AbstractSlice_createNewSliceFromString(
+    JNIEnv* env, jobject jobj, jstring str) {
+  const std::string s = rocksdb::JniUtil::copyString(env, str);
+  const rocksdb::Slice* slice = new rocksdb::Slice(s);
+  rocksdb::AbstractSliceJni::setHandle(env, jobj, slice);
+}
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    size0
+ * Signature: (J)I
+ */
+jint Java_org_rocksdb_AbstractSlice_size0(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  return slice->size();
+}
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    empty0
+ * Signature: (J)Z
+ */
+jboolean Java_org_rocksdb_AbstractSlice_empty0(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  return slice->empty();
+}
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    toString0
+ * Signature: (JZ)Ljava/lang/String;
+ */
+jstring Java_org_rocksdb_AbstractSlice_toString0(
+    JNIEnv* env, jobject jobj, jlong handle, jboolean hex) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  const std::string s = slice->ToString(hex);
+  return env->NewStringUTF(s.c_str());
+}
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    compare0
+ * Signature: (JJ)I;
+ */
+jint Java_org_rocksdb_AbstractSlice_compare0(
+    JNIEnv* env, jobject jobj, jlong handle, jlong otherHandle) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  const rocksdb::Slice* otherSlice =
+    reinterpret_cast<rocksdb::Slice*>(otherHandle);
+  return slice->compare(*otherSlice);
+}
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    startsWith0
+ * Signature: (JJ)Z;
+ */
+jboolean Java_org_rocksdb_AbstractSlice_startsWith0(
+    JNIEnv* env, jobject jobj, jlong handle, jlong otherHandle) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  const rocksdb::Slice* otherSlice =
+    reinterpret_cast<rocksdb::Slice*>(otherHandle);
+  return slice->starts_with(*otherSlice);
+}
+
+/*
+ * Class:     org_rocksdb_AbstractSlice
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_AbstractSlice_disposeInternal(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  delete reinterpret_cast<rocksdb::Slice*>(handle);
+}
+
+// </editor-fold>
+
+// <editor-fold desc="org.rocksdb.Slice>
+
+/*
+ * Class:     org_rocksdb_Slice
+ * Method:    createNewSlice0
+ * Signature: ([BI)V
+ */
+void Java_org_rocksdb_Slice_createNewSlice0(
+    JNIEnv * env, jobject jobj, jbyteArray data, jint offset) {
+
+  const jsize dataSize = env->GetArrayLength(data);
+  const int len = dataSize - offset;
+  jbyte* ptrData = new jbyte[len];
+  env->GetByteArrayRegion(data, offset, len, ptrData);
+
+  const rocksdb::Slice* slice = new rocksdb::Slice((const char*)ptrData, len);
+  rocksdb::AbstractSliceJni::setHandle(env, jobj, slice);
+}
+
+/*
+ * Class:     org_rocksdb_Slice
+ * Method:    createNewSlice1
+ * Signature: ([B)V
+ */
+void Java_org_rocksdb_Slice_createNewSlice1(
+    JNIEnv * env, jobject jobj, jbyteArray data) {
+
+  const int len = env->GetArrayLength(data);
+
+  jboolean isCopy;
+  jbyte* ptrData = env->GetByteArrayElements(data, &isCopy);
+  const char* buf = new char[len];
+  memcpy(const_cast<char*>(buf), ptrData, len);
+
+  const rocksdb::Slice* slice =
+    new rocksdb::Slice(buf, env->GetArrayLength(data));
+  rocksdb::AbstractSliceJni::setHandle(env, jobj, slice);
+
+  env->ReleaseByteArrayElements(data, ptrData, JNI_ABORT);
+
+  // NOTE: buf will be deleted in the org.rocksdb.Slice#dispose method
+}
+
+/*
+ * Class:     org_rocksdb_Slice
+ * Method:    data0
+ * Signature: (J)[B
+ */
+jbyteArray Java_org_rocksdb_Slice_data0(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  const int len = slice->size();
+  const jbyteArray data = env->NewByteArray(len);
+  env->SetByteArrayRegion(data, 0, len,
+    reinterpret_cast<jbyte*>(const_cast<char*>(slice->data())));
+  return data;
+}
+
+/*
+ * Class:     org_rocksdb_Slice
+ * Method:    disposeInternalBuf
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_Slice_disposeInternalBuf(
+    JNIEnv * env, jobject jobj, jlong handle) {
+    const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+    delete [] slice->data_;
+}
+
+// </editor-fold>
+
+// <editor-fold desc="org.rocksdb.DirectSlice>
+
+/*
+ * Class:     org_rocksdb_DirectSlice
+ * Method:    createNewDirectSlice0
+ * Signature: (Ljava/nio/ByteBuffer;I)V
+ */
+void Java_org_rocksdb_DirectSlice_createNewDirectSlice0(
+    JNIEnv* env, jobject jobj, jobject data, jint length) {
+  const char* ptrData =
+    reinterpret_cast<char*>(env->GetDirectBufferAddress(data));
+  const rocksdb::Slice* slice = new rocksdb::Slice(ptrData, length);
+  rocksdb::AbstractSliceJni::setHandle(env, jobj, slice);
+}
+
+/*
+ * Class:     org_rocksdb_DirectSlice
+ * Method:    createNewDirectSlice1
+ * Signature: (Ljava/nio/ByteBuffer;)V
+ */
+void Java_org_rocksdb_DirectSlice_createNewDirectSlice1(
+    JNIEnv* env, jobject jobj, jobject data) {
+  const char* ptrData =
+    reinterpret_cast<char*>(env->GetDirectBufferAddress(data));
+  const rocksdb::Slice* slice = new rocksdb::Slice(ptrData);
+  rocksdb::AbstractSliceJni::setHandle(env, jobj, slice);
+}
+
+/*
+ * Class:     org_rocksdb_DirectSlice
+ * Method:    data0
+ * Signature: (J)Ljava/lang/Object;
+ */
+jobject Java_org_rocksdb_DirectSlice_data0(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  const rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  return env->NewDirectByteBuffer(const_cast<char*>(slice->data()),
+    slice->size());
+}
+
+/*
+ * Class:     org_rocksdb_DirectSlice
+ * Method:    get0
+ * Signature: (JI)B
+ */
+jbyte Java_org_rocksdb_DirectSlice_get0(
+    JNIEnv* env, jobject jobj, jlong handle, jint offset) {
+  rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  return (*slice)[offset];
+}
+
+/*
+ * Class:     org_rocksdb_DirectSlice
+ * Method:    clear0
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_DirectSlice_clear0(
+    JNIEnv* env, jobject jobj, jlong handle) {
+  rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  delete [] slice->data_;
+  slice->clear();
+}
+
+/*
+ * Class:     org_rocksdb_DirectSlice
+ * Method:    removePrefix0
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_DirectSlice_removePrefix0(
+    JNIEnv* env, jobject jobj, jlong handle, jint length) {
+  rocksdb::Slice* slice = reinterpret_cast<rocksdb::Slice*>(handle);
+  slice->remove_prefix(length);
+}
+
+// </editor-fold>