refact

CMakeLists.txt

@@ -42,9 +42,10 @@ if(${BUILD_WITH_LEVELDB})
 endif(${BUILD_WITH_LEVELDB})
 
 ## rocksdb
-option(BUILD_WITH_ROCKSDB "If build with rocksdb" OFF)
+option(BUILD_WITH_ROCKSDB "If build with rocksdb" ON)
 if(${BUILD_WITH_ROCKSDB})
     cat("${CMAKE_SUPPORT_DIR}/rocksdb_CMakeLists.txt.in" ${DEPS_TMP_FILE})
+    add_definitions(-DUSE_ROCKSDB)
 endif(${BUILD_WITH_ROCKSDB})
 
 ## download dependencies

include/libs/tkv/tkv.h

@@ -32,13 +32,14 @@ typedef struct STkvWriteOpts STkvWriteOpts;
 // DB operations
 STkvDb *tkvOpen(const STkvOpts *options, const char *path);
 void    tkvClose(STkvDb *db);
-void    tkvPut(STkvDb *db, STkvWriteOpts *, char *key, size_t keylen, char *val, size_t vallen);
+void    tkvPut(STkvDb *db, const STkvWriteOpts *, const char *key, size_t keylen, const char *val, size_t vallen);
-char *  tkvGet(STkvDb *db, STkvReadOpts *, char *key, size_t keylen, size_t *vallen);
+char *  tkvGet(STkvDb *db, const STkvReadOpts *, const char *key, size_t keylen, size_t *vallen);
 
 // DB options
-STkvOpts *tkvOptionsCreate();
+STkvOpts *tkvOptsCreate();
-void      tkvOptionsDestroy(STkvOpts *);
+void      tkvOptsDestroy(STkvOpts *);
 void      tkvOptionsSetCache(STkvOpts *, STkvCache *);
+void      tkvOptsSetCreateIfMissing(STkvOpts *, unsigned char);
 
 // DB cache
 typedef enum { TKV_LRU_CACHE = 0, TKV_LFU_CACHE = 1 } ETkvCacheType;

source/libs/tkv/CMakeLists.txt

@@ -1,5 +1,5 @@
 aux_source_directory(src TKV_SRC)
-add_library(tkv ${TKV_SRC})
+add_library(tkv STATIC ${TKV_SRC})
 target_include_directories(
     tkv
     PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/tkv"

source/libs/tkv/src/tkv.c

@@ -13,69 +13,167 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#ifdef USE_ROCKSDB
+#include <rocksdb/c.h>
+#endif
+
 #include "tkv.h"
 
 struct STkvDb {
-  // TODO
+#ifdef USE_ROCKSDB
+  rocksdb_t *db;
+#endif
 };
+
 struct STkvOpts {
-  // TODO
+#ifdef USE_ROCKSDB
+  rocksdb_options_t *opts;
+#endif
 };
+
 struct STkvCache {
   // TODO
 };
+
 struct STkvReadOpts {
-  // TODO
+#ifdef USE_ROCKSDB
+  rocksdb_readoptions_t *ropts;
+#endif
 };
+
 struct STkvWriteOpts {
-  // TODO
+#ifdef USE_ROCKSDB
+  rocksdb_writeoptions_t *wopts;
+#endif
 };
 
 STkvDb *tkvOpen(const STkvOpts *options, const char *path) {
-  // TODO
+  STkvDb *pDb = NULL;
+
+  pDb = (STkvDb *)malloc(sizeof(*pDb));
+  if (pDb == NULL) {
     return NULL;
   }
 
-void tkvClose(STkvDb *db) {
+#ifdef USE_ROCKSDB
-  // TODO
+  char *err = NULL;
+
+  pDb->db = rocksdb_open(options->opts, path, &err);
+  // TODO: check err
+#endif
+
+  return pDb;
 }
 
-void tkvPut(STkvDb *db, STkvWriteOpts *pwopts, char *key, size_t keylen, char *val, size_t vallen) {
+void tkvClose(STkvDb *pDb) {
-  // TODO
+  if (pDb) {
+#ifdef USE_ROCKSDB
+    rocksdb_close(pDb->db);
+#endif
+    free(pDb);
+  }
 }
 
-char *tkvGet(STkvDb *db, STkvReadOpts *propts, char *key, size_t keylen, size_t *vallen) {
+void tkvPut(STkvDb *pDb, const STkvWriteOpts *pwopts, const char *key, size_t keylen, const char *val, size_t vallen) {
-  // TODO
+#ifdef USE_ROCKSDB
+  char *err = NULL;
+  rocksdb_put(pDb->db, pwopts->wopts, key, keylen, val, vallen, &err);
+  // TODO: check error
+#endif
+}
+
+char *tkvGet(STkvDb *pDb, const STkvReadOpts *propts, const char *key, size_t keylen, size_t *vallen) {
+  char *ret = NULL;
+
+#ifdef USE_ROCKSDB
+  char *err = NULL;
+  ret = rocksdb_get(pDb->db, propts->ropts, key, keylen, vallen, &err);
+  // TODD: check error
+#endif
+
+  return ret;
+}
+
+STkvOpts *tkvOptsCreate() {
+  STkvOpts *pOpts = NULL;
+
+  pOpts = (STkvOpts *)malloc(sizeof(*pOpts));
+  if (pOpts == NULL) {
     return NULL;
   }
 
-STkvOpts *tkvOptionsCreate() {
+#ifdef USE_ROCKSDB
-  // TODO
+  pOpts->opts = rocksdb_options_create();
-  return NULL;
+  // TODO: check error
+#endif
+
+  return pOpts;
 }
 
-void tkvOptionsDestroy(STkvOpts *popts) {
+void tkvOptsDestroy(STkvOpts *pOpts) {
-  // TODO
+  if (pOpts) {
+#ifdef USE_ROCKSDB
+    rocksdb_options_destroy(pOpts->opts);
+#endif
+    free(pOpts);
+  }
 }
 
 void tkvOptionsSetCache(STkvOpts *popts, STkvCache *pCache) {
   // TODO
 }
 
+void tkvOptsSetCreateIfMissing(STkvOpts *pOpts, unsigned char c) {
+#ifdef USE_ROCKSDB
+  rocksdb_options_set_create_if_missing(pOpts->opts, c);
+#endif
+}
+
 STkvReadOpts *tkvReadOptsCreate() {
-    // TODO
+  STkvReadOpts *pReadOpts = NULL;
+
+  pReadOpts = (STkvReadOpts *)malloc(sizeof(*pReadOpts));
+  if (pReadOpts == NULL) {
     return NULL;
   }
 
-void tkvReadOptsDestroy(STkvReadOpts *propts) {
+#ifdef USE_ROCKSDB
-  // TODO
+  pReadOpts->ropts = rocksdb_readoptions_create();
+#endif
+
+  return pReadOpts;
+}
+
+void tkvReadOptsDestroy(STkvReadOpts *pReadOpts) {
+  if (pReadOpts) {
+#ifdef USE_ROCKSDB
+    rocksdb_readoptions_destroy(pReadOpts->ropts);
+#endif
+    free(pReadOpts);
+  }
 }
 
 STkvWriteOpts *tkvWriteOptsCreate() {
-    // TODO
+  STkvWriteOpts *pWriteOpts = NULL;
+
+  pWriteOpts = (STkvWriteOpts *)malloc(sizeof(*pWriteOpts));
+  if (pWriteOpts == NULL) {
     return NULL;
   }
 
-void tkvWriteOptsDestroy(STkvWriteOpts *pwopts) {
+#ifdef USE_ROCKSDB
+  pWriteOpts->wopts = rocksdb_writeoptions_create();
+#endif
+
+  return pWriteOpts;
+}
+
+void tkvWriteOptsDestroy(STkvWriteOpts *pWriteOpts) {
+  if (pWriteOpts) {
+#ifdef USE_ROCKSDB
+    rocksdb_writeoptions_destroy(pWriteOpts->wopts);
+#endif
+    free(pWriteOpts);
+  }
   // TODO
 }

source/server/vnode/meta/src/meta.c

@@ -73,7 +73,8 @@ SMeta *metaOpen(SMetaOpts *options) {
   pMeta->stbList = tdListNew(sizeof(STableObj *));
 
   // Options
-  STkvOpts *dbOptions = tkvOptionsCreate();
+  STkvOpts *dbOptions = tkvOptsCreate();
+  tkvOptsSetCreateIfMissing(dbOptions, 1);
 
   taosMkDir("meta");
 
@@ -89,7 +90,7 @@ SMeta *metaOpen(SMetaOpts *options) {
   // Open tag index
   pMeta->tagIdx = tkvOpen(dbOptions, "meta/tag_idx_db");
 
-  tkvOptionsDestroy(dbOptions);
+  tkvOptsDestroy(dbOptions);
 
   return pMeta;
 }