Expose underlying Read/Write APIs for avoiding unnecessary memory copy (#2303)

Summary:
adamretter

As you already mentioned at #1247 .
Pull Request resolved: https://github.com/facebook/rocksdb/pull/2303

Differential Revision: D10209001

Pulled By: sagar0

fbshipit-source-id: bcbce004112c2edeaff116968d79c6f90aab4b6c

java/src/main/java/org/rocksdb/RocksDB.java

@@ -439,6 +439,12 @@ public class RocksDB extends RocksObject {
     options_ = options;
   }
 
+  private static void checkBounds(int offset, int len, int size) {
+    if ((offset | len | (offset + len) | (size - (offset + len))) < 0) {
+      throw new IndexOutOfBoundsException(String.format("offset(%d), len(%d), size(%d)", offset, len, size));
+    }
+  }
+
   /**
    * Set the database entry for "key" to "value".
    *
@@ -453,6 +459,28 @@ public class RocksDB extends RocksObject {
     put(nativeHandle_, key, 0, key.length, value, 0, value.length);
   }
 
+  /**
+   * Set the database entry for "key" to "value"
+   *
+   * @param key The specified key to be inserted
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("key".length -  offset)
+   * @param value the value associated with the specified key
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if errors happens in underlying native library.
+   */
+  public void put(final byte[] key, int offset, int len, final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    put(nativeHandle_, key, offset, len, value, vOffset, vLen);
+  }
+
   /**
    * Set the database entry for "key" to "value" in the specified
    * column family.
@@ -473,6 +501,32 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Set the database entry for "key" to "value" in the specified
+   * column family.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param key The specified key to be inserted
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("key".length -  offset)
+   * @param value the value associated with the specified key
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if errors happens in underlying native library.
+   */
+  public void put(final ColumnFamilyHandle columnFamilyHandle, final byte[] key, int offset, int len, final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    put(nativeHandle_, key, offset, len, value, vOffset, vLen,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * Set the database entry for "key" to "value".
    *
@@ -489,6 +543,32 @@ public class RocksDB extends RocksObject {
         key, 0, key.length, value, 0, value.length);
   }
 
+  /**
+   * Set the database entry for "key" to "value".
+   *
+   * @param writeOpts {@link org.rocksdb.WriteOptions} instance.
+   * @param key The specified key to be inserted
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("key".length -  offset)
+   * @param value the value associated with the specified key
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void put(final WriteOptions writeOpts, byte[] key, int offset, int len, byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    put(nativeHandle_, writeOpts.nativeHandle_,
+        key, offset, len, value, vOffset, vLen);
+  }
+
+
   /**
    * Set the database entry for "key" to "value" for the specified
    * column family.
@@ -512,6 +592,36 @@ public class RocksDB extends RocksObject {
         0, value.length, columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Set the database entry for "key" to "value" for the specified
+   * column family.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param writeOpts {@link org.rocksdb.WriteOptions} instance.
+   * @param key The specified key to be inserted
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("key".length -  offset)
+   * @param value the value associated with the specified key
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void put(final ColumnFamilyHandle columnFamilyHandle,
+      final WriteOptions writeOpts, final byte[] key, int offset, int len,
+      final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    put(nativeHandle_, writeOpts.nativeHandle_, key, offset, len, value,
+        vOffset, vLen, columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * If the key definitely does not exist in the database, then this method
    * returns false, else true.
@@ -528,6 +638,27 @@ public class RocksDB extends RocksObject {
     return keyMayExist(nativeHandle_, key, 0, key.length, value);
   }
 
+  /**
+   * If the key definitely does not exist in the database, then this method
+   * returns false, else true.
+   *
+   * This check is potentially lighter-weight than invoking DB::Get(). One way
+   * to make this lighter weight is to avoid doing any IOs.
+   *
+   * @param key byte array of a key to search for
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value StringBuilder instance which is a out parameter if a value is
+   *    found in block-cache.
+   *
+   * @return boolean value indicating if key does not exist or might exist.
+   */
+  public boolean keyMayExist(final byte[] key, int offset, int len, final StringBuilder value) {
+    checkBounds(offset, len, key.length);
+    return keyMayExist(nativeHandle_, key, offset, len, value);
+  }
+
   /**
    * If the key definitely does not exist in the database, then this method
    * returns false, else true.
@@ -547,6 +678,30 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_, value);
   }
 
+  /**
+   * If the key definitely does not exist in the database, then this method
+   * returns false, else true.
+   *
+   * This check is potentially lighter-weight than invoking DB::Get(). One way
+   * to make this lighter weight is to avoid doing any IOs.
+   *
+   * @param columnFamilyHandle {@link ColumnFamilyHandle} instance
+   * @param key byte array of a key to search for
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value StringBuilder instance which is a out parameter if a value is
+   *    found in block-cache.
+   * @return boolean value indicating if key does not exist or might exist.
+   */
+  public boolean keyMayExist(final ColumnFamilyHandle columnFamilyHandle,
+      final byte[] key, int offset, int len, final StringBuilder value) {
+    checkBounds(offset, len, key.length);
+    return keyMayExist(nativeHandle_, key, offset, len,
+        columnFamilyHandle.nativeHandle_, value);
+  }
+
+
   /**
    * If the key definitely does not exist in the database, then this method
    * returns false, else true.
@@ -566,6 +721,29 @@ public class RocksDB extends RocksObject {
         key, 0, key.length, value);
   }
 
+  /**
+   * If the key definitely does not exist in the database, then this method
+   * returns false, else true.
+   *
+   * This check is potentially lighter-weight than invoking DB::Get(). One way
+   * to make this lighter weight is to avoid doing any IOs.
+   *
+   * @param readOptions {@link ReadOptions} instance
+   * @param key byte array of a key to search for
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value StringBuilder instance which is a out parameter if a value is
+   *    found in block-cache.
+   * @return boolean value indicating if key does not exist or might exist.
+   */
+  public boolean keyMayExist(final ReadOptions readOptions,
+      final byte[] key, int offset, int len, final StringBuilder value) {
+    checkBounds(offset, len, key.length);
+    return keyMayExist(nativeHandle_, readOptions.nativeHandle_,
+        key, offset, len, value);
+  }
+
   /**
    * If the key definitely does not exist in the database, then this method
    * returns false, else true.
@@ -588,6 +766,32 @@ public class RocksDB extends RocksObject {
         value);
   }
 
+  /**
+   * If the key definitely does not exist in the database, then this method
+   * returns false, else true.
+   *
+   * This check is potentially lighter-weight than invoking DB::Get(). One way
+   * to make this lighter weight is to avoid doing any IOs.
+   *
+   * @param readOptions {@link ReadOptions} instance
+   * @param columnFamilyHandle {@link ColumnFamilyHandle} instance
+   * @param key byte array of a key to search for
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value StringBuilder instance which is a out parameter if a value is
+   *    found in block-cache.
+   * @return boolean value indicating if key does not exist or might exist.
+   */
+  public boolean keyMayExist(final ReadOptions readOptions,
+      final ColumnFamilyHandle columnFamilyHandle, final byte[] key, int offset, int len,
+      final StringBuilder value) {
+    checkBounds(offset, len, key.length);
+    return keyMayExist(nativeHandle_, readOptions.nativeHandle_,
+        key, offset, len, columnFamilyHandle.nativeHandle_,
+        value);
+  }
+
   /**
    * Apply the specified updates to the database.
    *
@@ -631,6 +835,30 @@ public class RocksDB extends RocksObject {
     merge(nativeHandle_, key, 0, key.length, value, 0, value.length);
   }
 
+  /**
+   * Add merge operand for key/value pair.
+   *
+   * @param key the specified key to be merged.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the value to be merged with the current value for the specified key.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void merge(final byte[] key, int offset, int len, final byte[] value, int vOffset, int vLen)
+      throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    merge(nativeHandle_, key, offset, len, value, vOffset, vLen);
+  }
+
+
   /**
    * Add merge operand for key/value pair in a ColumnFamily.
    *
@@ -648,6 +876,32 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Add merge operand for key/value pair in a ColumnFamily.
+   *
+   * @param columnFamilyHandle {@link ColumnFamilyHandle} instance
+   * @param key the specified key to be merged.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the value to be merged with the current value for
+   * the specified key.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void merge(final ColumnFamilyHandle columnFamilyHandle,
+      final byte[] key, int offset, int len, final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    merge(nativeHandle_, key, offset, len, value, vOffset, vLen,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * Add merge operand for key/value pair.
    *
@@ -665,6 +919,32 @@ public class RocksDB extends RocksObject {
         key, 0, key.length, value, 0, value.length);
   }
 
+  /**
+   * Add merge operand for key/value pair.
+   *
+   * @param writeOpts {@link WriteOptions} for this write.
+   * @param key the specified key to be merged.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the value to be merged with the current value for
+   * the specified key.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void merge(final WriteOptions writeOpts, final byte[] key, int offset, int len,
+      final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    merge(nativeHandle_, writeOpts.nativeHandle_,
+        key, offset, len, value, vOffset, vLen);
+  }
+
   /**
    * Add merge operand for key/value pair.
    *
@@ -685,13 +965,44 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Add merge operand for key/value pair.
+   *
+   * @param columnFamilyHandle {@link ColumnFamilyHandle} instance
+   * @param writeOpts {@link WriteOptions} for this write.
+   * @param key the specified key to be merged.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the value to be merged with the current value for
+   * the specified key.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void merge(final ColumnFamilyHandle columnFamilyHandle,
+      final WriteOptions writeOpts, final byte[] key, int offset, int len,
+      final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    merge(nativeHandle_, writeOpts.nativeHandle_,
+        key, offset, len, value, vOffset, vLen,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   // TODO(AR) we should improve the #get() API, returning -1 (RocksDB.NOT_FOUND) is not very nice
   // when we could communicate better status into, also the C++ code show that -2 could be returned
 
   /**
    * Get the value associated with the specified key within column family*
+   *
    * @param key the key to retrieve the value.
    * @param value the out-value to receive the retrieved value.
+   *
    * @return The size of the actual value that matches the specified
    *     {@code key} in byte.  If the return value is greater than the
    *     length of {@code value}, then it indicates that the size of the
@@ -706,6 +1017,35 @@ public class RocksDB extends RocksObject {
     return get(nativeHandle_, key, 0, key.length, value, 0, value.length);
   }
 
+  /**
+   * Get the value associated with the specified key within column family*
+   *
+   * @param key the key to retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the out-value to receive the retrieved value.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @return The size of the actual value that matches the specified
+   *     {@code key} in byte.  If the return value is greater than the
+   *     length of {@code value}, then it indicates that the size of the
+   *     input buffer {@code value} is insufficient and partial result will
+   *     be returned.  RocksDB.NOT_FOUND will be returned if the value not
+   *     found.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public int get(final byte[] key, int offset, int len, final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    return get(nativeHandle_, key, offset, len, value, vOffset, vLen);
+  }
+
   /**
    * Get the value associated with the specified key within column family.
    *
@@ -729,6 +1069,39 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Get the value associated with the specified key within column family.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param key the key to retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the out-value to receive the retrieved value.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   *
+   * @return The size of the actual value that matches the specified
+   *     {@code key} in byte.  If the return value is greater than the
+   *     length of {@code value}, then it indicates that the size of the
+   *     input buffer {@code value} is insufficient and partial result will
+   *     be returned.  RocksDB.NOT_FOUND will be returned if the value not
+   *     found.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public int get(final ColumnFamilyHandle columnFamilyHandle, final byte[] key, int offset, int len,
+      final byte[] value, int vOffset, int vLen) throws RocksDBException, IllegalArgumentException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    return get(nativeHandle_, key, offset, len, value, vOffset, vLen,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * Get the value associated with the specified key.
    *
@@ -750,6 +1123,38 @@ public class RocksDB extends RocksObject {
     return get(nativeHandle_, opt.nativeHandle_,
                key, 0, key.length, value, 0, value.length);
   }
+
+  /**
+   * Get the value associated with the specified key.
+   *
+   * @param opt {@link org.rocksdb.ReadOptions} instance.
+   * @param key the key to retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the out-value to receive the retrieved value.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   * @return The size of the actual value that matches the specified
+   *     {@code key} in byte.  If the return value is greater than the
+   *     length of {@code value}, then it indicates that the size of the
+   *     input buffer {@code value} is insufficient and partial result will
+   *     be returned.  RocksDB.NOT_FOUND will be returned if the value not
+   *     found.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public int get(final ReadOptions opt, final byte[] key, int offset, int len,
+      final byte[] value, int vOffset, int vLen) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    return get(nativeHandle_, opt.nativeHandle_,
+        key, offset, len, value, vOffset, vLen);
+  }
+
   /**
    * Get the value associated with the specified key within column family.
    *
@@ -775,6 +1180,40 @@ public class RocksDB extends RocksObject {
         0, value.length, columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Get the value associated with the specified key within column family.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param opt {@link org.rocksdb.ReadOptions} instance.
+   * @param key the key to retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param value the out-value to receive the retrieved value.
+   * @param vOffset the offset of the "value" array to be used, must be non-negative and
+   *               no longer than "key".length
+   * @param vLen the length of the "value" array to be used, must be non-negative and
+   *            must be non-negative and no larger than ("value".length -  offset)
+   * @return The size of the actual value that matches the specified
+   *     {@code key} in byte.  If the return value is greater than the
+   *     length of {@code value}, then it indicates that the size of the
+   *     input buffer {@code value} is insufficient and partial result will
+   *     be returned.  RocksDB.NOT_FOUND will be returned if the value not
+   *     found.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public int get(final ColumnFamilyHandle columnFamilyHandle,
+      final ReadOptions opt, final byte[] key, int offset, int len, final byte[] value, int vOffset, int vLen)
+      throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    checkBounds(vOffset, vLen, value.length);
+    return get(nativeHandle_, opt.nativeHandle_, key, offset, len, value,
+        vOffset, vLen, columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * The simplified version of get which returns a new byte array storing
    * the value associated with the specified input key if any.  null will be
@@ -791,6 +1230,26 @@ public class RocksDB extends RocksObject {
     return get(nativeHandle_, key, 0, key.length);
   }
 
+  /**
+   * The simplified version of get which returns a new byte array storing
+   * the value associated with the specified input key if any.  null will be
+   * returned if the specified key is not found.
+   *
+   * @param key the key retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @return a byte array storing the value associated with the input key if
+   *     any.  null if it does not find the specified key.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public byte[] get(final byte[] key, int offset, int len) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    return get(nativeHandle_, key, offset, len);
+  }
+
   /**
    * The simplified version of get which returns a new byte array storing
    * the value associated with the specified input key if any.  null will be
@@ -811,6 +1270,30 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * The simplified version of get which returns a new byte array storing
+   * the value associated with the specified input key if any.  null will be
+   * returned if the specified key is not found.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param key the key retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @return a byte array storing the value associated with the input key if
+   *     any.  null if it does not find the specified key.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public byte[] get(final ColumnFamilyHandle columnFamilyHandle,
+      final byte[] key, int offset, int len) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    return get(nativeHandle_, key, offset, len,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * The simplified version of get which returns a new byte array storing
    * the value associated with the specified input key if any.  null will be
@@ -829,6 +1312,28 @@ public class RocksDB extends RocksObject {
     return get(nativeHandle_, opt.nativeHandle_, key, 0, key.length);
   }
 
+  /**
+   * The simplified version of get which returns a new byte array storing
+   * the value associated with the specified input key if any.  null will be
+   * returned if the specified key is not found.
+   *
+   * @param key the key retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param opt Read options.
+   * @return a byte array storing the value associated with the input key if
+   *     any.  null if it does not find the specified key.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public byte[] get(final ReadOptions opt, final byte[] key, int offset, int len)
+      throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    return get(nativeHandle_, opt.nativeHandle_, key, offset, len);
+  }
+
   /**
    * The simplified version of get which returns a new byte array storing
    * the value associated with the specified input key if any.  null will be
@@ -850,6 +1355,31 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * The simplified version of get which returns a new byte array storing
+   * the value associated with the specified input key if any.  null will be
+   * returned if the specified key is not found.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param key the key retrieve the value.
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   * @param opt Read options.
+   * @return a byte array storing the value associated with the input key if
+   *     any.  null if it does not find the specified key.
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public byte[] get(final ColumnFamilyHandle columnFamilyHandle,
+      final ReadOptions opt, final byte[] key, int offset, int len) throws RocksDBException {
+    checkBounds(offset, len, key.length);
+    return get(nativeHandle_, opt.nativeHandle_, key, offset, len,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * Returns a map of keys for which values were found in DB.
    *
@@ -1073,6 +1603,23 @@ public class RocksDB extends RocksObject {
     delete(nativeHandle_, key, 0, key.length);
   }
 
+  /**
+   * Delete the database entry (if any) for "key".  Returns OK on
+   * success, and a non-OK status on error.  It is not an error if "key"
+   * did not exist in the database.
+   *
+   * @param key Key to delete within database
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void delete(final byte[] key, int offset, int len) throws RocksDBException {
+    delete(nativeHandle_, key, offset, len);
+  }
+
   /**
    * Remove the database entry (if any) for "key".  Returns OK on
    * success, and a non-OK status on error.  It is not an error if "key"
@@ -1110,6 +1657,26 @@ public class RocksDB extends RocksObject {
     delete(nativeHandle_, key, 0, key.length, columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Delete the database entry (if any) for "key".  Returns OK on
+   * success, and a non-OK status on error.  It is not an error if "key"
+   * did not exist in the database.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param key Key to delete within database
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void delete(final ColumnFamilyHandle columnFamilyHandle,
+      final byte[] key, int offset, int len) throws RocksDBException {
+    delete(nativeHandle_, key, offset, len, columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * Remove the database entry (if any) for "key".  Returns OK on
    * success, and a non-OK status on error.  It is not an error if "key"
@@ -1145,6 +1712,25 @@ public class RocksDB extends RocksObject {
     delete(nativeHandle_, writeOpt.nativeHandle_, key, 0, key.length);
   }
 
+  /**
+   * Delete the database entry (if any) for "key".  Returns OK on
+   * success, and a non-OK status on error.  It is not an error if "key"
+   * did not exist in the database.
+   *
+   * @param writeOpt WriteOptions to be used with delete operation
+   * @param key Key to delete within database
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void delete(final WriteOptions writeOpt, final byte[] key, int offset, int len)
+      throws RocksDBException {
+    delete(nativeHandle_, writeOpt.nativeHandle_, key, offset, len);
+  }
+
   /**
    * Remove the database entry (if any) for "key".  Returns OK on
    * success, and a non-OK status on error.  It is not an error if "key"
@@ -1187,6 +1773,29 @@ public class RocksDB extends RocksObject {
         columnFamilyHandle.nativeHandle_);
   }
 
+  /**
+   * Delete the database entry (if any) for "key".  Returns OK on
+   * success, and a non-OK status on error.  It is not an error if "key"
+   * did not exist in the database.
+   *
+   * @param columnFamilyHandle {@link org.rocksdb.ColumnFamilyHandle}
+   *     instance
+   * @param writeOpt WriteOptions to be used with delete operation
+   * @param key Key to delete within database
+   * @param offset the offset of the "key" array to be used, must be non-negative and
+   *               no larger than "key".length
+   * @param len the length of the "key" array to be used, must be non-negative and
+   *
+   * @throws RocksDBException thrown if error happens in underlying
+   *    native library.
+   */
+  public void delete(final ColumnFamilyHandle columnFamilyHandle,
+      final WriteOptions writeOpt, final byte[] key, int offset, int len)
+      throws RocksDBException {
+    delete(nativeHandle_, writeOpt.nativeHandle_, key, offset, len,
+        columnFamilyHandle.nativeHandle_);
+  }
+
   /**
    * Remove the database entry for {@code key}. Requires that the key exists
    * and was not overwritten. It is not an error if the key did not exist

java/src/test/java/org/rocksdb/KeyMayExistTest.java

@@ -48,12 +48,33 @@ public class KeyMayExistTest {
         assertThat(exists).isTrue();
         assertThat(retValue.toString()).isEqualTo("value");
 
+        // Slice key
+        StringBuilder builder = new StringBuilder("prefix");
+        int offset = builder.toString().length();
+        builder.append("slice key 0");
+        int len = builder.toString().length() - offset;
+        builder.append("suffix");
+
+        byte[] sliceKey = builder.toString().getBytes();
+        byte[] sliceValue = "slice value 0".getBytes();
+        db.put(sliceKey, offset, len, sliceValue, 0, sliceValue.length);
+
+        retValue = new StringBuilder();
+        exists = db.keyMayExist(sliceKey, offset, len, retValue);
+        assertThat(exists).isTrue();
+        assertThat(retValue.toString().getBytes()).isEqualTo(sliceValue);
+
         // Test without column family but with readOptions
         try (final ReadOptions readOptions = new ReadOptions()) {
           retValue = new StringBuilder();
           exists = db.keyMayExist(readOptions, "key".getBytes(), retValue);
           assertThat(exists).isTrue();
           assertThat(retValue.toString()).isEqualTo("value");
+
+          retValue = new StringBuilder();
+          exists = db.keyMayExist(readOptions, sliceKey, offset, len, retValue);
+          assertThat(exists).isTrue();
+          assertThat(retValue.toString().getBytes()).isEqualTo(sliceValue);
         }
 
         // Test with column family
@@ -63,6 +84,13 @@ public class KeyMayExistTest {
         assertThat(exists).isTrue();
         assertThat(retValue.toString()).isEqualTo("value");
 
+        // Test slice sky with column family
+        retValue = new StringBuilder();
+        exists = db.keyMayExist(columnFamilyHandleList.get(0), sliceKey, offset, len,
+            retValue);
+        assertThat(exists).isTrue();
+        assertThat(retValue.toString().getBytes()).isEqualTo(sliceValue);
+
         // Test with column family and readOptions
         try (final ReadOptions readOptions = new ReadOptions()) {
           retValue = new StringBuilder();
@@ -71,11 +99,23 @@ public class KeyMayExistTest {
               retValue);
           assertThat(exists).isTrue();
           assertThat(retValue.toString()).isEqualTo("value");
+
+          // Test slice key with column family and read options
+          retValue = new StringBuilder();
+          exists = db.keyMayExist(readOptions,
+              columnFamilyHandleList.get(0), sliceKey, offset, len,
+              retValue);
+          assertThat(exists).isTrue();
+          assertThat(retValue.toString().getBytes()).isEqualTo(sliceValue);
         }
 
         // KeyMayExist in CF1 must return false
         assertThat(db.keyMayExist(columnFamilyHandleList.get(1),
             "key".getBytes(), retValue)).isFalse();
+
+        // slice key
+        assertThat(db.keyMayExist(columnFamilyHandleList.get(1),
+           sliceKey, 1, 3, retValue)).isFalse();
       } finally {
         for (final ColumnFamilyHandle columnFamilyHandle :
             columnFamilyHandleList) {

java/src/test/java/org/rocksdb/RocksDBTest.java

@@ -4,6 +4,7 @@
 //  (found in the LICENSE.Apache file in the root directory).
 package org.rocksdb;
 
+import org.junit.Assert;
 import org.junit.Assume;
 import org.junit.ClassRule;
 import org.junit.Rule;
@@ -11,6 +12,7 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 import org.junit.rules.TemporaryFolder;
 
+import java.nio.ByteBuffer;
 import java.util.*;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -70,6 +72,57 @@ public class RocksDBTest {
           "value".getBytes());
       assertThat(db.get("key2".getBytes())).isEqualTo(
           "12345678".getBytes());
+
+
+      // put
+      Segment key3 = sliceSegment("key3");
+      Segment key4 = sliceSegment("key4");
+      Segment value0 = sliceSegment("value 0");
+      Segment value1 = sliceSegment("value 1");
+      db.put(key3.data, key3.offset, key3.len, value0.data, value0.offset, value0.len);
+      db.put(opt, key4.data, key4.offset, key4.len, value1.data, value1.offset, value1.len);
+
+      // compare
+      Assert.assertTrue(value0.isSamePayload(db.get(key3.data, key3.offset, key3.len)));
+      Assert.assertTrue(value1.isSamePayload(db.get(key4.data, key4.offset, key4.len)));
+    }
+  }
+
+  private static Segment sliceSegment(String key) {
+    ByteBuffer rawKey = ByteBuffer.allocate(key.length() + 4);
+    rawKey.put((byte)0);
+    rawKey.put((byte)0);
+    rawKey.put(key.getBytes());
+
+    return new Segment(rawKey.array(), 2, key.length());
+  }
+
+  private static class Segment {
+    final byte[] data;
+    final int offset;
+    final int len;
+
+    public boolean isSamePayload(byte[] value) {
+      if (value == null) {
+        return false;
+      }
+      if (value.length != len) {
+        return false;
+      }
+
+      for (int i = 0; i < value.length; i++) {
+        if (data[i + offset] != value[i]) {
+          return false;
+        }
+      }
+
+      return true;
+    }
+
+    public Segment(byte[] value, int offset, int len) {
+      this.data = value;
+      this.offset = offset;
+      this.len = len;
     }
   }
 
@@ -242,6 +295,18 @@ public class RocksDBTest {
       db.merge(wOpt, "key2".getBytes(), "xxxx".getBytes());
       assertThat(db.get("key2".getBytes())).isEqualTo(
           "xxxx".getBytes());
+
+      Segment key3 = sliceSegment("key3");
+      Segment key4 = sliceSegment("key4");
+      Segment value0 = sliceSegment("value 0");
+      Segment value1 = sliceSegment("value 1");
+
+      db.merge(key3.data, key3.offset, key3.len, value0.data, value0.offset, value0.len);
+      db.merge(wOpt, key4.data, key4.offset, key4.len, value1.data, value1.offset, value1.len);
+
+      // compare
+      Assert.assertTrue(value0.isSamePayload(db.get(key3.data, key3.offset, key3.len)));
+      Assert.assertTrue(value1.isSamePayload(db.get(key4.data, key4.offset, key4.len)));
     }
   }
 
@@ -259,6 +324,18 @@ public class RocksDBTest {
       db.delete(wOpt, "key2".getBytes());
       assertThat(db.get("key1".getBytes())).isNull();
       assertThat(db.get("key2".getBytes())).isNull();
+
+
+      Segment key3 = sliceSegment("key3");
+      Segment key4 = sliceSegment("key4");
+      db.put("key3".getBytes(), "key3 value".getBytes());
+      db.put("key4".getBytes(), "key4 value".getBytes());
+
+      db.delete(key3.data, key3.offset, key3.len);
+      db.delete(wOpt, key4.data, key4.offset, key4.len);
+
+      assertThat(db.get("key3".getBytes())).isNull();
+      assertThat(db.get("key4".getBytes())).isNull();
     }
   }