Fix create_cf not passinig titan specific options to titan (#311)

Summary:
`db.create_cf` call `crocksdb_create_column_family`, which doesn't pass titan specific options to titan. Fix it by calling `ctitandb_create_column_family` when titan is enabled.

Test Plan:
Manual test when enabled titan, tikv pass correct titan options when creating new CFs.

Will add unit test in later patch.
Signed-off-by: Yi Wu <yiwu@pingcap.com>

librocksdb_sys/crocksdb/c.cc

@@ -4958,6 +4958,9 @@ struct ctitandb_options_t {
   TitanOptions rep;
 };
 
+// TODO: Simplify the API by merging db_options into tdb_options, and
+// column_family_options into titan_column_family_options, since the later
+// of the pairs already contain the former.
 crocksdb_t* ctitandb_open_column_families(
     const char* name, const crocksdb_options_t* db_options,
     const ctitandb_options_t* tdb_options, int num_column_families,
@@ -4993,6 +4996,26 @@ crocksdb_t* ctitandb_open_column_families(
   return result;
 }
 
+// Caller should make sure `db` is created from ctitandb_open_column_families.
+//
+// TODO: ctitandb_open_column_family should return a ctitandb_t. Caller can
+// use ctitandb_t for titan specific functions.
+crocksdb_column_family_handle_t* ctitandb_create_column_family(
+    crocksdb_t* db,
+    const ctitandb_options_t* titan_column_family_options,
+    const char* column_family_name,
+    char** errptr) {
+  // Blindly cast db into TitanDB.
+  TitanDB* titan_db = reinterpret_cast<TitanDB*>(db->rep);
+  crocksdb_column_family_handle_t* handle = new crocksdb_column_family_handle_t;
+  SaveError(errptr,
+      titan_db->CreateColumnFamily(
+          TitanCFDescriptor(std::string(column_family_name),
+                            titan_column_family_options->rep),
+          &(handle->rep)));
+  return handle;
+}
+
 /* TitanDBOptions */
 
 ctitandb_options_t* ctitandb_options_create() { return new ctitandb_options_t; }

librocksdb_sys/crocksdb/crocksdb/c.h

@@ -1976,6 +1976,13 @@ extern C_ROCKSDB_LIBRARY_API crocksdb_t* ctitandb_open_column_families(
     const ctitandb_options_t** titan_column_family_options,
     crocksdb_column_family_handle_t** column_family_handles, char** errptr);
 
+extern C_ROCKSDB_LIBRARY_API
+crocksdb_column_family_handle_t* ctitandb_create_column_family(
+    crocksdb_t* db,
+    const ctitandb_options_t* titan_column_family_options,
+    const char* column_family_name,
+    char** errptr);
+
 /* TitanDBOptions */
 
 extern C_ROCKSDB_LIBRARY_API ctitandb_options_t* ctitandb_options_create();

librocksdb_sys/src/lib.rs

@@ -1826,6 +1826,13 @@ extern "C" {
         err: *mut *mut c_char,
     ) -> *mut DBInstance;
 
+    pub fn ctitandb_create_column_family(
+        db: *mut DBInstance,
+        titan_column_family_options: *const DBTitanDBOptions,
+        column_family_name: *const c_char,
+        err: *mut *mut c_char,
+    ) -> *mut DBCFHandle;
+
     pub fn ctitandb_options_create() -> *mut DBTitanDBOptions;
     pub fn ctitandb_options_destroy(opts: *mut DBTitanDBOptions);
     pub fn ctitandb_options_copy(opts: *mut DBTitanDBOptions) -> *mut DBTitanDBOptions;

src/rocksdb.rs

@@ -771,11 +771,19 @@ impl DB {
         };
         let cname_ptr = cname.as_ptr();
         unsafe {
-            let cf_handler = ffi_try!(crocksdb_create_column_family(
-                self.inner,
-                cfd.options.inner,
-                cname_ptr
-            ));
+            let cf_handler = if !self.is_titan() {
+                ffi_try!(crocksdb_create_column_family(
+                    self.inner,
+                    cfd.options.inner,
+                    cname_ptr
+                ))
+            } else {
+                ffi_try!(ctitandb_create_column_family(
+                    self.inner,
+                    cfd.options.titan_inner,
+                    cname_ptr
+                ))
+            };
             let handle = CFHandle { inner: cf_handler };
             self._cf_opts.push(cfd.options);
             Ok(match self.cfs.entry(cfd.name.to_owned()) {