[RocksJava] ColumnFamily custom Options API extension

*********************
                   ***************************
                 ******** ************* ********
                ********   ***********   ********
               ********     *********     ********
              *************************************
              *************************************
              *************************************
               ******     ***       ***     ******
                ******    ***  ***  ***    ******
                 ******        ***        ******
                   ***************************
                      *********************

java/Makefile

@@ -5,6 +5,7 @@ NATIVE_JAVA_CLASSES = org.rocksdb.AbstractComparator\
 	org.rocksdb.BlockBasedTableConfig\
 	org.rocksdb.BloomFilter\
 	org.rocksdb.ColumnFamilyHandle\
+	org.rocksdb.ColumnFamilyOptions\
 	org.rocksdb.Comparator\
 	org.rocksdb.ComparatorOptions\
 	org.rocksdb.DBOptions\

java/org/rocksdb/ColumnFamilyDescriptor.java

@@ -0,0 +1,58 @@
+// 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;
+
+/**
+ * <p>Describes a column family with a
+ * name and respective Options.</p>
+ */
+public class ColumnFamilyDescriptor {
+
+  /**
+   * <p>Creates a new Column Family using a name and default
+   * options,</p>
+   *
+   * @param columnFamilyName name of column family.
+   */
+  public ColumnFamilyDescriptor(final String columnFamilyName){
+    this(columnFamilyName, new ColumnFamilyOptions());
+  }
+
+  /**
+   * <p>Creates a new Column Family using a name and custom
+   * options.</p>
+   *
+   * @param columnFamilyName name of column family.
+   * @param columnFamilyOptions options to be used with
+   *     column family.
+   */
+  public ColumnFamilyDescriptor(final String columnFamilyName,
+      final ColumnFamilyOptions columnFamilyOptions) {
+    columnFamilyName_ = columnFamilyName;
+    columnFamilyOptions_ = columnFamilyOptions;
+  }
+
+  /**
+   * Retrieve name of column family.
+   *
+   * @return column family name.
+   */
+  public String columnFamilyName() {
+    return columnFamilyName_;
+  }
+
+  /**
+   * Retrieve assigned options instance.
+   *
+   * @return Options instance assigned to this instance.
+   */
+  public ColumnFamilyOptions columnFamilyOptions() {
+    return columnFamilyOptions_;
+  }
+
+  private final String columnFamilyName_;
+  private final ColumnFamilyOptions columnFamilyOptions_;
+}

java/org/rocksdb/RocksDB.java

@@ -123,7 +123,7 @@ public class RocksDB extends RocksObject {
    * </p>
    *
    * @param path the path to the rocksdb.
-   * @param columnFamilyNames list of column family names
+   * @param columnFamilyDescriptors list of column family descriptors
    * @param columnFamilyHandles will be filled with ColumnFamilyHandle instances
    *     on open.
    * @return a {@link RocksDB} instance on success, null if the specified
@@ -132,12 +132,13 @@ public class RocksDB extends RocksObject {
    * @throws org.rocksdb.RocksDBException
    * @see Options#setCreateIfMissing(boolean)
    */
-  public static RocksDB open(String path, List<String> columnFamilyNames,
+  public static RocksDB open(String path,
+      List<ColumnFamilyDescriptor> columnFamilyDescriptors,
       List<ColumnFamilyHandle> columnFamilyHandles) throws RocksDBException {
     // This allows to use the rocksjni default Options instead of
     // the c++ one.
     Options options = new Options();
-    return open(options, path, columnFamilyNames, columnFamilyHandles);
+    return open(options, path, columnFamilyDescriptors, columnFamilyHandles);
   }
 
   /**
@@ -198,7 +199,7 @@ public class RocksDB extends RocksObject {
    *
    * @param options {@link org.rocksdb.Options} instance.
    * @param path the path to the rocksdb.
-   * @param columnFamilyNames list of column family names
+   * @param columnFamilyDescriptors list of column family descriptors
    * @param columnFamilyHandles will be filled with ColumnFamilyHandle instances
    *     on open.
    * @return a {@link RocksDB} instance on success, null if the specified
@@ -207,13 +208,14 @@ public class RocksDB extends RocksObject {
    * @throws org.rocksdb.RocksDBException
    * @see Options#setCreateIfMissing(boolean)
    */
-  public static RocksDB open(Options options, String path, List<String> columnFamilyNames,
+  public static RocksDB open(Options options, String path,
+      List<ColumnFamilyDescriptor> columnFamilyDescriptors,
       List<ColumnFamilyHandle> columnFamilyHandles)
       throws RocksDBException {
     RocksDB db = new RocksDB();
     List<Long> cfReferences = db.open(options.nativeHandle_, path,
-        columnFamilyNames, columnFamilyNames.size());
-    for (int i=0; i<columnFamilyNames.size(); i++) {
+        columnFamilyDescriptors, columnFamilyDescriptors.size());
+    for (int i=0; i<columnFamilyDescriptors.size(); i++) {
       columnFamilyHandles.add(new ColumnFamilyHandle(db, cfReferences.get(i)));
     }
     db.storeOptionsInstance(options);
@@ -244,19 +246,21 @@ public class RocksDB extends RocksObject {
    * options.
    *
    * @param path the path to the RocksDB.
-   * @param columnFamilyNames list of column family names
+   * @param columnFamilyDescriptors list of column family descriptors
    * @param columnFamilyHandles will be filled with ColumnFamilyHandle instances
    *     on open.
    * @return a {@link RocksDB} instance on success, null if the specified
    *     {@link RocksDB} can not be opened.
    * @throws RocksDBException
    */
-  public static RocksDB openReadOnly(String path, List<String> columnFamilyNames,
+  public static RocksDB openReadOnly(String path,
+      List<ColumnFamilyDescriptor> columnFamilyDescriptors,
       List<ColumnFamilyHandle> columnFamilyHandles) throws RocksDBException {
     // This allows to use the rocksjni default Options instead of
     // the c++ one.
     Options options = new Options();
-    return openReadOnly(options, path, columnFamilyNames, columnFamilyHandles);
+    return openReadOnly(options, path, columnFamilyDescriptors,
+        columnFamilyHandles);
   }
 
   /**
@@ -299,7 +303,7 @@ public class RocksDB extends RocksObject {
    *
    * @param options {@link Options} instance.
    * @param path the path to the RocksDB.
-   * @param columnFamilyNames list of column family names
+   * @param columnFamilyDescriptors list of column family descriptors
    * @param columnFamilyHandles will be filled with ColumnFamilyHandle instances
    *     on open.
    * @return a {@link RocksDB} instance on success, null if the specified
@@ -307,15 +311,16 @@ public class RocksDB extends RocksObject {
    * @throws RocksDBException
    */
   public static RocksDB openReadOnly(Options options, String path,
-      List<String> columnFamilyNames, List<ColumnFamilyHandle> columnFamilyHandles)
+      List<ColumnFamilyDescriptor> columnFamilyDescriptors,
+      List<ColumnFamilyHandle> columnFamilyHandles)
       throws RocksDBException {
     // when non-default Options is used, keeping an Options reference
     // in RocksDB can prevent Java to GC during the life-time of
     // the currently-created RocksDB.
     RocksDB db = new RocksDB();
     List<Long> cfReferences = db.openROnly(options.nativeHandle_, path,
-        columnFamilyNames, columnFamilyNames.size());
-    for (int i=0; i<columnFamilyNames.size(); i++) {
+        columnFamilyDescriptors, columnFamilyDescriptors.size());
+    for (int i=0; i<columnFamilyDescriptors.size(); i++) {
       columnFamilyHandles.add(new ColumnFamilyHandle(db, cfReferences.get(i)));
     }
 
@@ -1059,14 +1064,15 @@ public class RocksDB extends RocksObject {
    * allocates a ColumnFamilyHandle within an internal structure.
    * The ColumnFamilyHandle is automatically disposed with DB disposal.
    *
-   * @param columnFamilyName Name of column family to be created.
+   * @param columnFamilyDescriptor column family to be created.
    * @return {@link org.rocksdb.ColumnFamilyHandle} instance
    * @see RocksDBException
    */
-  public ColumnFamilyHandle createColumnFamily(String columnFamilyName)
+  public ColumnFamilyHandle createColumnFamily(
+      ColumnFamilyDescriptor columnFamilyDescriptor)
       throws RocksDBException {
     return new ColumnFamilyHandle(this, createColumnFamily(nativeHandle_,
-        options_.nativeHandle_, columnFamilyName));
+        columnFamilyDescriptor));
   }
 
   /**
@@ -1130,15 +1136,17 @@ public class RocksDB extends RocksObject {
   protected native void open(
       long optionsHandle, String path) throws RocksDBException;
   protected native List<Long> open(long optionsHandle, String path,
-      List<String> columnFamilyNames, int columnFamilyNamesLength)
+      List<ColumnFamilyDescriptor> columnFamilyDescriptors,
+      int columnFamilyDescriptorsLength)
       throws RocksDBException;
   protected native static List<byte[]> listColumnFamilies(
       long optionsHandle, String path) throws RocksDBException;
   protected native void openROnly(
       long optionsHandle, String path) throws RocksDBException;
   protected native List<Long> openROnly(
-      long optionsHandle, String path, List<String> columnFamilyNames,
-      int columnFamilyNamesLength) throws RocksDBException;
+      long optionsHandle, String path,
+      List<ColumnFamilyDescriptor> columnFamilyDescriptors,
+      int columnFamilyDescriptorsLength) throws RocksDBException;
   protected native void put(
       long handle, byte[] key, int keyLen,
       byte[] value, int valueLen) throws RocksDBException;
@@ -1231,8 +1239,8 @@ public class RocksDB extends RocksObject {
   protected native void releaseSnapshot(
       long nativeHandle, long snapshotHandle);
   private native void disposeInternal(long handle);
-  private native long createColumnFamily(long handle, long opt_handle,
-      String name) throws RocksDBException;
+  private native long createColumnFamily(long handle,
+      ColumnFamilyDescriptor columnFamilyDescriptor) throws RocksDBException;
   private native void dropColumnFamily(long handle, long cfHandle) throws RocksDBException;
   private native void flush(long handle, long flushOptHandle)
       throws RocksDBException;

java/org/rocksdb/test/ColumnFamilyTest.java

@@ -43,7 +43,8 @@ public class ColumnFamilyTest {
 
     // Test createColumnFamily
     try {
-      db.createColumnFamily("new_cf");
+      db.createColumnFamily(new ColumnFamilyDescriptor("new_cf",
+          new ColumnFamilyOptions()));
     } catch (RocksDBException e) {
       assert(false);
     }
@@ -67,11 +68,12 @@ public class ColumnFamilyTest {
     }
 
     // Test open database with column family names
-    List<String> cfNames = new ArrayList<String>();
+    List<ColumnFamilyDescriptor> cfNames =
+        new ArrayList<ColumnFamilyDescriptor>();
     List<ColumnFamilyHandle> columnFamilyHandleList =
         new ArrayList<ColumnFamilyHandle>();
-    cfNames.add("default");
-    cfNames.add("new_cf");
+    cfNames.add(new ColumnFamilyDescriptor("default"));
+    cfNames.add(new ColumnFamilyDescriptor("new_cf"));
 
     try {
       db = RocksDB.open(options, db_path, cfNames, columnFamilyHandleList);
@@ -100,7 +102,8 @@ public class ColumnFamilyTest {
     // Test create write to and drop ColumnFamily
     ColumnFamilyHandle tmpColumnFamilyHandle = null;
     try {
-      tmpColumnFamilyHandle = db.createColumnFamily("tmpCF");
+      tmpColumnFamilyHandle = db.createColumnFamily(
+          new ColumnFamilyDescriptor("tmpCF", new ColumnFamilyOptions()));
       db.put(tmpColumnFamilyHandle, "key".getBytes(), "value".getBytes());
       db.dropColumnFamily(tmpColumnFamilyHandle);
       tmpColumnFamilyHandle.dispose();

java/org/rocksdb/test/KeyMayExistTest.java

@@ -22,11 +22,12 @@ public class KeyMayExistTest {
         .setCreateMissingColumnFamilies(true);
     try {
       // open database using cf names
-      List<String> cfNames = new ArrayList<String>();
+      List<ColumnFamilyDescriptor> cfNames =
+          new ArrayList<ColumnFamilyDescriptor>();
       List<ColumnFamilyHandle> columnFamilyHandleList =
           new ArrayList<ColumnFamilyHandle>();
-      cfNames.add("default");
-      cfNames.add("new_cf");
+      cfNames.add(new ColumnFamilyDescriptor("default"));
+      cfNames.add(new ColumnFamilyDescriptor("new_cf"));
       db = RocksDB.open(options, DB_PATH, cfNames, columnFamilyHandleList);
       assert(columnFamilyHandleList.size()==2);
 

java/org/rocksdb/test/MergeTest.java

@@ -46,13 +46,18 @@ public class MergeTest {
     opt.setCreateMissingColumnFamilies(true);
     opt.setMergeOperatorName("stringappend");
 
-    List<String> cfNames = new ArrayList<String>();
+    List<ColumnFamilyDescriptor> cfDescr =
+        new ArrayList<ColumnFamilyDescriptor>();
     List<ColumnFamilyHandle> columnFamilyHandleList =
     new ArrayList<ColumnFamilyHandle>();
-    cfNames.add("default");
-    cfNames.add("new_cf");
+    cfDescr.add(new ColumnFamilyDescriptor("default",
+        new ColumnFamilyOptions().setMergeOperatorName(
+            "stringappend")));
+    cfDescr.add(new ColumnFamilyDescriptor("default",
+        new ColumnFamilyOptions().setMergeOperatorName(
+            "stringappend")));
     RocksDB db = RocksDB.open(opt, db_cf_path_string,
-        cfNames, columnFamilyHandleList);
+        cfDescr, columnFamilyHandleList);
 
     // writing aa under key
     db.put(columnFamilyHandleList.get(1),
@@ -103,13 +108,18 @@ public class MergeTest {
     StringAppendOperator stringAppendOperator = new StringAppendOperator();
     opt.setMergeOperator(stringAppendOperator);
 
-    List<String> cfNames = new ArrayList<String>();
+    List<ColumnFamilyDescriptor> cfDescr =
+        new ArrayList<ColumnFamilyDescriptor>();
     List<ColumnFamilyHandle> columnFamilyHandleList =
     new ArrayList<ColumnFamilyHandle>();
-    cfNames.add("default");
-    cfNames.add("new_cf");
+    cfDescr.add(new ColumnFamilyDescriptor("default",
+        new ColumnFamilyOptions().setMergeOperator(
+            stringAppendOperator)));
+    cfDescr.add(new ColumnFamilyDescriptor("new_cf",
+        new ColumnFamilyOptions().setMergeOperator(
+            stringAppendOperator)));
     RocksDB db = RocksDB.open(opt, db_path_operator,
-        cfNames, columnFamilyHandleList);
+        cfDescr, columnFamilyHandleList);
 
     // writing aa under key
     db.put(columnFamilyHandleList.get(1),
@@ -121,7 +131,10 @@ public class MergeTest {
     String strValue = new String(value);
 
     // Test also with createColumnFamily
-    ColumnFamilyHandle columnFamilyHandle = db.createColumnFamily("new_cf2");
+    ColumnFamilyHandle columnFamilyHandle = db.createColumnFamily(
+        new ColumnFamilyDescriptor("new_cf2",
+            new ColumnFamilyOptions().setMergeOperator(
+                new StringAppendOperator())));
     // writing xx under cfkey2
     db.put(columnFamilyHandle, "cfkey2".getBytes(), "xx".getBytes());
     // merge yy under cfkey2

java/org/rocksdb/test/ReadOnlyTest.java

@@ -34,12 +34,15 @@ public class ReadOnlyTest {
       db2.close();
 
 
-      List<String> cfNames = new ArrayList<String>();
-      cfNames.add("default");
+      List<ColumnFamilyDescriptor> cfNames =
+          new ArrayList<ColumnFamilyDescriptor>();
+      cfNames.add(new ColumnFamilyDescriptor("default"));
 
       db = RocksDB.open(DB_PATH, cfNames, columnFamilyHandleList);
-      columnFamilyHandleList.add(db.createColumnFamily("new_cf"));
-      columnFamilyHandleList.add(db.createColumnFamily("new_cf2"));
+      columnFamilyHandleList.add(db.createColumnFamily(
+          new ColumnFamilyDescriptor("new_cf", new ColumnFamilyOptions())));
+      columnFamilyHandleList.add(db.createColumnFamily(
+          new ColumnFamilyDescriptor("new_cf2", new ColumnFamilyOptions())));
       db.put(columnFamilyHandleList.get(2), "key2".getBytes(),
           "value2".getBytes());
 
@@ -47,9 +50,10 @@ public class ReadOnlyTest {
       assert(db2.get("key2".getBytes())==null);
       assert(db2.get(columnFamilyHandleList.get(0), "key2".getBytes())==null);
 
-      List<String> cfNewName = new ArrayList<String>();
-      cfNewName.add("default");
-      cfNewName.add("new_cf2");
+      List<ColumnFamilyDescriptor> cfNewName =
+          new ArrayList<ColumnFamilyDescriptor>();
+      cfNewName.add(new ColumnFamilyDescriptor("default"));
+      cfNewName.add(new ColumnFamilyDescriptor("new_cf2"));
       db3 = RocksDB.openReadOnly(DB_PATH, cfNewName, db3ColumnFamilyHandleList);
       assert(new String(db3.get(db3ColumnFamilyHandleList.get(1),
           "key2".getBytes())).equals("value2"));

java/rocksjni/options.cc

@@ -12,9 +12,8 @@
 #include <memory>
 
 #include "include/org_rocksdb_Options.h"
-//TODO(fyrz) to be commented in with options refactoring pull requests
 #include "include/org_rocksdb_DBOptions.h"
-//#include "include/org_rocksdb_ColumnFamilyOptions.h"
+#include "include/org_rocksdb_ColumnFamilyOptions.h"
 #include "include/org_rocksdb_WriteOptions.h"
 #include "include/org_rocksdb_ReadOptions.h"
 #include "include/org_rocksdb_ComparatorOptions.h"

java/rocksjni/portal.h

@@ -159,6 +159,34 @@ class DBOptionsJni {
   }
 };
 
+class ColumnFamilyDescriptorJni {
+ public:
+  // Get the java class id of org.rocksdb.ColumnFamilyDescriptor
+  static jclass getColumnFamilyDescriptorClass(JNIEnv* env) {
+    jclass jclazz = env->FindClass("org/rocksdb/ColumnFamilyDescriptor");
+    assert(jclazz != nullptr);
+    return jclazz;
+  }
+
+  // Get the java method id of columnFamilyName
+  static jmethodID getColumnFamilyNameMethod(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(
+        getColumnFamilyDescriptorClass(env),
+        "columnFamilyName", "()Ljava/lang/String;");
+    assert(mid != nullptr);
+    return mid;
+  }
+
+  // Get the java method id of columnFamilyOptions
+  static jmethodID getColumnFamilyOptionsMethod(JNIEnv* env) {
+    static jmethodID mid = env->GetMethodID(
+        getColumnFamilyDescriptorClass(env),
+        "columnFamilyOptions", "()Lorg/rocksdb/ColumnFamilyOptions;");
+    assert(mid != nullptr);
+    return mid;
+  }
+};
+
 class ColumnFamilyOptionsJni {
  public:
   // Get the java class id of org.rocksdb.ColumnFamilyOptions.

java/rocksjni/rocksjni.cc

@@ -69,7 +69,7 @@ void Java_org_rocksdb_RocksDB_openROnly__JLjava_lang_String_2(
 jobject
     Java_org_rocksdb_RocksDB_openROnly__JLjava_lang_String_2Ljava_util_List_2I(
     JNIEnv* env, jobject jdb, jlong jopt_handle, jstring jdb_path,
-    jobject jcfname_list, jint jcfname_count) {
+    jobject jcfdesc_list, jint jcfdesc_count) {
   auto opt = reinterpret_cast<rocksdb::Options*>(jopt_handle);
   rocksdb::DB* db = nullptr;
   const char* db_path = env->GetStringUTFChars(jdb_path, 0);
@@ -79,23 +79,34 @@ jobject
 
   std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
   std::vector<rocksdb::ColumnFamilyHandle* > handles;
-  // get iterator for cfnames
+  // get iterator for ColumnFamilyDescriptors
   jobject iteratorObj = env->CallObjectMethod(
-      jcfname_list, rocksdb::ListJni::getIteratorMethod(env));
+      jcfdesc_list, rocksdb::ListJni::getIteratorMethod(env));
 
-  // iterate over cfnames and convert cfnames to
-  // ColumnFamilyDescriptor instances
+  // iterate over ColumnFamilyDescriptors
   while (env->CallBooleanMethod(
       iteratorObj, rocksdb::ListJni::getHasNextMethod(env)) == JNI_TRUE) {
-      jstring jstr = (jstring) env->CallObjectMethod(iteratorObj,
+      // get ColumnFamilyDescriptor
+      jobject jcf_descriptor = env->CallObjectMethod(iteratorObj,
           rocksdb::ListJni::getNextMethod(env));
+      // get ColumnFamilyName
+      jstring jstr = (jstring) env->CallObjectMethod(jcf_descriptor,
+          rocksdb::ColumnFamilyDescriptorJni::getColumnFamilyNameMethod(
+          env));
+      // get CF Options
+      jobject jcf_opt_obj = env->CallObjectMethod(jcf_descriptor,
+          rocksdb::ColumnFamilyDescriptorJni::getColumnFamilyOptionsMethod(
+          env));
+      rocksdb::ColumnFamilyOptions* cfOptions =
+          rocksdb::ColumnFamilyOptionsJni::getHandle(env, jcf_opt_obj);
+
       const char* cfname = env->GetStringUTFChars(jstr, 0);
 
       // free allocated cfnames after call to open
       cfnames_to_free.push_back(cfname);
       jcfnames_for_free.push_back(jstr);
       column_families.push_back(rocksdb::ColumnFamilyDescriptor(cfname,
-          *static_cast<rocksdb::ColumnFamilyOptions*>(opt)));
+          *cfOptions));
   }
 
   rocksdb::Status s = rocksdb::DB::OpenForReadOnly(*opt,
@@ -141,7 +152,7 @@ jobject
  */
 jobject Java_org_rocksdb_RocksDB_open__JLjava_lang_String_2Ljava_util_List_2I(
     JNIEnv* env, jobject jdb, jlong jopt_handle, jstring jdb_path,
-    jobject jcfname_list, jint jcfname_count) {
+    jobject jcfdesc_list, jint jcfdesc_count) {
   auto opt = reinterpret_cast<rocksdb::Options*>(jopt_handle);
   rocksdb::DB* db = nullptr;
   const char* db_path = env->GetStringUTFChars(jdb_path, 0);
@@ -151,23 +162,34 @@ jobject Java_org_rocksdb_RocksDB_open__JLjava_lang_String_2Ljava_util_List_2I(
 
   std::vector<rocksdb::ColumnFamilyDescriptor> column_families;
   std::vector<rocksdb::ColumnFamilyHandle* > handles;
-  // get iterator for cfnames
+  // get iterator for ColumnFamilyDescriptors
   jobject iteratorObj = env->CallObjectMethod(
-      jcfname_list, rocksdb::ListJni::getIteratorMethod(env));
+      jcfdesc_list, rocksdb::ListJni::getIteratorMethod(env));
 
-  // iterate over cfnames and convert cfnames to
-  // ColumnFamilyDescriptor instances
+  // iterate over ColumnFamilyDescriptors
   while (env->CallBooleanMethod(
       iteratorObj, rocksdb::ListJni::getHasNextMethod(env)) == JNI_TRUE) {
-      jstring jstr = (jstring) env->CallObjectMethod(iteratorObj,
+      // get ColumnFamilyDescriptor
+      jobject jcf_descriptor = env->CallObjectMethod(iteratorObj,
           rocksdb::ListJni::getNextMethod(env));
+      // get ColumnFamilyName
+      jstring jstr = (jstring) env->CallObjectMethod(jcf_descriptor,
+          rocksdb::ColumnFamilyDescriptorJni::getColumnFamilyNameMethod(
+          env));
+      // get CF Options
+      jobject jcf_opt_obj = env->CallObjectMethod(jcf_descriptor,
+          rocksdb::ColumnFamilyDescriptorJni::getColumnFamilyOptionsMethod(
+          env));
+      rocksdb::ColumnFamilyOptions* cfOptions =
+          rocksdb::ColumnFamilyOptionsJni::getHandle(env, jcf_opt_obj);
+
       const char* cfname = env->GetStringUTFChars(jstr, 0);
 
       // free allocated cfnames after call to open
       cfnames_to_free.push_back(cfname);
       jcfnames_for_free.push_back(jstr);
       column_families.push_back(rocksdb::ColumnFamilyDescriptor(cfname,
-          *static_cast<rocksdb::ColumnFamilyOptions*>(opt)));
+          *cfOptions));
   }
 
   rocksdb::Status s = rocksdb::DB::Open(*opt, db_path, column_families,
@@ -1151,18 +1173,28 @@ jlongArray Java_org_rocksdb_RocksDB_iterators(
 /*
  * Class:     org_rocksdb_RocksDB
  * Method:    createColumnFamily
- * Signature: (JJLjava/lang/String;)J;
+ * Signature: (JLorg/rocksdb/ColumnFamilyDescriptor;)J;
  */
 jlong Java_org_rocksdb_RocksDB_createColumnFamily(
-    JNIEnv* env, jobject jdb, jlong jdb_handle, jlong jopt_handle,
-    jstring jcfname) {
+    JNIEnv* env, jobject jdb, jlong jdb_handle,
+    jobject jcf_descriptor) {
   rocksdb::ColumnFamilyHandle* handle;
-  const char* cfname = env->GetStringUTFChars(jcfname, 0);
   auto db_handle = reinterpret_cast<rocksdb::DB*>(jdb_handle);
-  auto opt = reinterpret_cast<rocksdb::Options*>(jopt_handle);
+
+  jstring jstr = (jstring) env->CallObjectMethod(jcf_descriptor,
+      rocksdb::ColumnFamilyDescriptorJni::getColumnFamilyNameMethod(
+      env));
+  // get CF Options
+  jobject jcf_opt_obj = env->CallObjectMethod(jcf_descriptor,
+      rocksdb::ColumnFamilyDescriptorJni::getColumnFamilyOptionsMethod(
+      env));
+  rocksdb::ColumnFamilyOptions* cfOptions =
+      rocksdb::ColumnFamilyOptionsJni::getHandle(env, jcf_opt_obj);
+
+  const char* cfname = env->GetStringUTFChars(jstr, 0);
   rocksdb::Status s = db_handle->CreateColumnFamily(
-      *static_cast<rocksdb::ColumnFamilyOptions*>(opt), cfname, &handle);
-  env->ReleaseStringUTFChars(jcfname, cfname);
+      *cfOptions, cfname, &handle);
+  env->ReleaseStringUTFChars(jstr, cfname);
 
   if (s.ok()) {
     return reinterpret_cast<jlong>(handle);