Merge pull request #12113 from taosdata/feature/vnode_refact1

feat: drop table and tdb upsert

include/util/tskiplist2.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+#ifndef _TD_UTIL_SKIPLIST2_H_
+#define _TD_UTIL_SKIPLIST2_H_
+
+#include "os.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define SL_MAX_LEVEL 15
+
+typedef struct SSkipList2 SSkipList2;
+typedef struct SSLCursor  SSLCursor;
+typedef struct SSLCfg     SSLCfg;
+typedef struct SSLNode    SSLNode;
+
+typedef int32_t (*tslCmprFn)(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2);
+
+// SSkipList2
+int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl);
+int32_t slClose(SSkipList2 *pSl);
+int32_t slClear(SSkipList2 *pSl);
+
+// SSLCursor
+int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc);
+int32_t slcClose(SSLCursor *pSlc);
+int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey);
+int32_t slcMoveToNext(SSLCursor *pSlc);
+int32_t slcMoveToPrev(SSLCursor *pSlc);
+int32_t slcMoveToFirst(SSLCursor *pSlc);
+int32_t slcMoveToLast(SSLCursor *pSlc);
+int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData);
+int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData);
+int32_t slcDrop(SSLCursor *pSlc);
+
+// struct
+struct SSLCfg {
+  int8_t    maxLevel;
+  int32_t   nKey;
+  int32_t   nData;
+  tslCmprFn cmprFn;
+  void     *pPool;
+  void *(*xMalloc)(void *, int32_t size);
+  void (*xFree)(void *, void *);
+};
+
+struct SSLCursor {
+  SSkipList2 *pSl;
+  SSLNode   **forwards[SL_MAX_LEVEL];
+};
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_UTIL_SKIPLIST2_H_*/
\ No newline at end of file

source/dnode/vnode/src/inc/meta.h

@@ -103,8 +103,6 @@ typedef struct {
 
 #if 1
 
-// int             metaCreateTable(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle);
-int          metaDropTable(SMeta* pMeta, tb_uid_t uid);
 SMSmaCursor* metaOpenSmaCursor(SMeta* pMeta, tb_uid_t uid);
 void         metaCloseSmaCursor(SMSmaCursor* pSmaCur);
 int64_t      metaSmaCursorNext(SMSmaCursor* pSmaCur);

source/dnode/vnode/src/inc/vnodeInt.h

@@ -46,13 +46,13 @@
 extern "C" {
 #endif
 
-typedef struct SVnodeInfo   SVnodeInfo;
-typedef struct SMeta        SMeta;
-typedef struct STsdb        STsdb;
-typedef struct STQ          STQ;
-typedef struct SVState      SVState;
-typedef struct SVBufPool    SVBufPool;
-typedef struct SQWorker SQHandle;
+typedef struct SVnodeInfo SVnodeInfo;
+typedef struct SMeta      SMeta;
+typedef struct STsdb      STsdb;
+typedef struct STQ        STQ;
+typedef struct SVState    SVState;
+typedef struct SVBufPool  SVBufPool;
+typedef struct SQWorker   SQHandle;
 
 #define VNODE_META_DIR  "meta"
 #define VNODE_TSDB_DIR  "tsdb"
@@ -77,6 +77,7 @@ int             metaCommit(SMeta* pMeta);
 int             metaCreateSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
 int             metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq);
 int             metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq);
+int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq);
 SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline);
 STSchema*       metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
 int             metaGetTableEntryByName(SMetaReader* pReader, const char* name);
@@ -100,7 +101,7 @@ int32_t      tsdbCreateTSma(STsdb* pTsdb, char* pMsg);
 int32_t      tsdbInsertTSmaData(STsdb* pTsdb, int64_t indexUid, const char* msg);
 int          tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
-                              uint64_t taskId);
+                             uint64_t taskId);
 tsdbReaderT  tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
                                  void* pMemRef);
 int32_t      tsdbGetTableGroupFromIdListT(STsdb* tsdb, SArray* pTableIdList, STableGroupInfo* pGroupInfo);
@@ -189,7 +190,6 @@ struct STbUidStore {
 
 #define TD_VID(PVNODE) (PVNODE)->config.vgId
 
-
 static FORCE_INLINE bool vnodeIsRollup(SVnode* pVnode) {
   SRetention* pRetention = &(pVnode->config.tsdbCfg.retentions[0]);
   return (pRetention->freq > 0 && pRetention->keep > 0);

source/dnode/vnode/src/meta/metaTDBImpl.c

@@ -289,7 +289,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
   pVal = pBuf = buf;
   metaEncodeTbInfo(&pBuf, pTbCfg);
   vLen = POINTER_DISTANCE(pBuf, buf);
-  ret = tdbDbPut(pMetaDb->pTbDB, pKey, kLen, pVal, vLen, &pMetaDb->txn);
+  ret = tdbDbInsert(pMetaDb->pTbDB, pKey, kLen, pVal, vLen, &pMetaDb->txn);
   if (ret < 0) {
     return -1;
   }
@@ -311,7 +311,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
     pVal = pBuf = buf;
     metaEncodeSchemaEx(&pBuf, &schemaWrapper);
     vLen = POINTER_DISTANCE(pBuf, buf);
-    ret = tdbDbPut(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen, &pMeta->pDB->txn);
+    ret = tdbDbInsert(pMetaDb->pSchemaDB, pKey, kLen, pVal, vLen, &pMeta->pDB->txn);
     if (ret < 0) {
       return -1;
     }
@@ -325,7 +325,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
   kLen = nameLen + 1 + sizeof(uid);
   pVal = NULL;
   vLen = 0;
-  ret = tdbDbPut(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
+  ret = tdbDbInsert(pMetaDb->pNameIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
   if (ret < 0) {
     return -1;
   }
@@ -336,7 +336,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
     kLen = sizeof(uid);
     pVal = NULL;
     vLen = 0;
-    ret = tdbDbPut(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
+    ret = tdbDbInsert(pMetaDb->pStbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
     if (ret < 0) {
       return -1;
     }
@@ -347,7 +347,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
     kLen = sizeof(ctbIdxKey);
     pVal = NULL;
     vLen = 0;
-    ret = tdbDbPut(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
+    ret = tdbDbInsert(pMetaDb->pCtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
     if (ret < 0) {
       return -1;
     }
@@ -362,7 +362,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
     kLen = sizeof(uid);
     pVal = NULL;
     vLen = 0;
-    ret = tdbDbPut(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
+    ret = tdbDbInsert(pMetaDb->pNtbIdx, pKey, kLen, pVal, vLen, &pMetaDb->txn);
     if (ret < 0) {
       return -1;
     }
@@ -530,7 +530,7 @@ int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
   int32_t kLen = sizeof(pSmaCfg->indexUid);
   int32_t vLen = POINTER_DISTANCE(qBuf, pBuf);
 
-  ret = tdbDbPut(pMeta->pDB->pSmaDB, key, kLen, val, vLen, &pMetaDb->txn);
+  ret = tdbDbInsert(pMeta->pDB->pSmaDB, key, kLen, val, vLen, &pMetaDb->txn);
   if (ret < 0) {
     taosMemoryFreeClear(pBuf);
     return -1;
@@ -545,7 +545,7 @@ int metaSaveSmaToDB(SMeta *pMeta, STSma *pSmaCfg) {
   val = NULL;
   vLen = 0;
 
-  ret = tdbDbPut(pMeta->pDB->pSmaIdx, key, kLen, val, vLen, &pMetaDb->txn);
+  ret = tdbDbInsert(pMeta->pDB->pSmaIdx, key, kLen, val, vLen, &pMetaDb->txn);
   if (ret < 0) {
     taosMemoryFreeClear(pBuf);
     return -1;

source/dnode/vnode/src/meta/metaTable.c

@@ -72,44 +72,61 @@ _err:
 }
 
 int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
-  SMetaReader mr = {0};
-
-  // validate req
-  metaReaderInit(&mr, pMeta, 0);
-  if (metaGetTableEntryByUid(&mr, pReq->suid) < 0) {
-    terrno = TSDB_CODE_VND_TABLE_NOT_EXIST;
+  TDBC       *pNameIdxc = NULL;
+  TDBC       *pUidIdxc = NULL;
+  TDBC       *pCtbIdxc = NULL;
+  SCtbIdxKey *pCtbIdxKey;
+  const void *pKey = NULL;
+  int         nKey;
+  const void *pData = NULL;
+  int         nData;
+  int         c, ret;
+
+  // prepare uid idx cursor
+  tdbDbcOpen(pMeta->pUidIdx, &pUidIdxc, &pMeta->txn);
+  ret = tdbDbcMoveTo(pUidIdxc, &pReq->suid, sizeof(tb_uid_t), &c);
+  if (ret < 0 || c != 0) {
+    terrno = TSDB_CODE_VND_TB_NOT_EXIST;
+    tdbDbcClose(pUidIdxc);
     goto _err;
   }
 
-  // do drop
-  // drop from pTbDb
-  // drop from pSkmDb
-  // drop from pUidIdx
-  // drop from pNameIdx
-  // {
-  //   TDBC *pDbc1 = NULL;
-  //   void *pKey = NULL;
-  //   void *pVal = NULL;
-  //   int   kLen = 0;
-  //   int   vLen = 0;
-  //   int   ret = 0;
-
-  //   // drop from pCtbIdx
-  //   ret = tdbDbcOpen(pMeta->pCtbIdx, &pDbc1);
-  //   tdbDbcMoveTo(pDbc1, &pReq->suid, sizeof(pReq->suid), NULL /*cmpr*/, 0 /*TDB_FORWARD_SEARCH*/);
-  //   tdbDbcGet(pDbc1, &pKey, &kLen, &pVal, vLen);
-  //   tdbDbcDrop(pDbc1);
-  //   // drop from pTagIdx
-  //   // drop from pTtlIdx
-  // }
-
-  // clear and return
-  metaReaderClear(&mr);
-  metaError("vgId:%d  super table %s uid:%" PRId64 " is dropped", TD_VID(pMeta->pVnode), pReq->name, pReq->suid);
+  // prepare name idx cursor
+  tdbDbcOpen(pMeta->pNameIdx, &pNameIdxc, &pMeta->txn);
+  ret = tdbDbcMoveTo(pNameIdxc, pReq->name, strlen(pReq->name) + 1, &c);
+  if (ret < 0 || c != 0) {
+    ASSERT(0);
+  }
+
+  tdbDbcDelete(pUidIdxc);
+  tdbDbcDelete(pNameIdxc);
+  tdbDbcClose(pUidIdxc);
+  tdbDbcClose(pNameIdxc);
+
+  // loop to drop each child table
+  tdbDbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
+  ret = tdbDbcMoveTo(pCtbIdxc, &(SCtbIdxKey){.suid = pReq->suid, .uid = INT64_MIN}, sizeof(SCtbIdxKey), &c);
+  if (ret < 0 || (c < 0 && tdbDbcMoveToNext(pCtbIdxc) < 0)) {
+    tdbDbcClose(pCtbIdxc);
+    goto _exit;
+  }
+
+  for (;;) {
+    tdbDbcGet(pCtbIdxc, &pKey, &nKey, NULL, NULL);
+    pCtbIdxKey = (SCtbIdxKey *)pKey;
+
+    if (pCtbIdxKey->suid > pReq->suid) break;
+
+    // drop the child table (TODO)
+
+    if (tdbDbcMoveToNext(pCtbIdxc) < 0) break;
+  }
+
+_exit:
+  metaDebug("vgId:%d  super table %s uid:%" PRId64 " is dropped", TD_VID(pMeta->pVnode), pReq->name, pReq->suid);
   return 0;
 
 _err:
-  metaReaderClear(&mr);
   metaError("vgId:%d failed to drop super table %s uid:%" PRId64 " since %s", TD_VID(pMeta->pVnode), pReq->name,
             pReq->suid, tstrerror(terrno));
   return -1;
@@ -166,18 +183,122 @@ _err:
   return -1;
 }
 
-int metaDropTable(SMeta *pMeta, tb_uid_t uid) {
-#if 0
-  if (metaRemoveTableFromIdx(pMeta, uid) < 0) {
-    // TODO: handle error
+int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
+  TDBC       *pTbDbc = NULL;
+  TDBC       *pUidIdxc = NULL;
+  TDBC       *pNameIdxc = NULL;
+  const void *pData;
+  int         nData;
+  tb_uid_t    uid;
+  int64_t     tver;
+  SMetaEntry  me = {0};
+  SCoder      coder = {0};
+  int8_t      type;
+  int64_t     ctime;
+  tb_uid_t    suid;
+  int         c, ret;
+
+  // search & delete the name idx
+  tdbDbcOpen(pMeta->pNameIdx, &pNameIdxc, &pMeta->txn);
+  ret = tdbDbcMoveTo(pNameIdxc, pReq->name, strlen(pReq->name) + 1, &c);
+  if (ret < 0 || c) {
+    tdbDbcClose(pNameIdxc);
+    terrno = TSDB_CODE_VND_TABLE_NOT_EXIST;
     return -1;
   }
 
-  if (metaRemoveTableFromIdx(pMeta, uid) < 0) {
-    // TODO
+  ret = tdbDbcGet(pNameIdxc, NULL, NULL, &pData, &nData);
+  if (ret < 0) {
+    ASSERT(0);
     return -1;
   }
-#endif
+
+  uid = *(tb_uid_t *)pData;
+
+  tdbDbcDelete(pNameIdxc);
+  tdbDbcClose(pNameIdxc);
+
+  // search & delete uid idx
+  tdbDbcOpen(pMeta->pUidIdx, &pUidIdxc, &pMeta->txn);
+  ret = tdbDbcMoveTo(pUidIdxc, &uid, sizeof(uid), &c);
+  if (ret < 0 || c != 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  ret = tdbDbcGet(pUidIdxc, NULL, NULL, &pData, &nData);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  tver = *(int64_t *)pData;
+  tdbDbcDelete(pUidIdxc);
+  tdbDbcClose(pUidIdxc);
+
+  // search and get meta entry
+  tdbDbcOpen(pMeta->pTbDb, &pTbDbc, &pMeta->txn);
+  ret = tdbDbcMoveTo(pTbDbc, &(STbDbKey){.uid = uid, .version = tver}, sizeof(STbDbKey), &c);
+  if (ret < 0 || c != 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  ret = tdbDbcGet(pTbDbc, NULL, NULL, &pData, &nData);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  // decode entry
+  void *pDataCopy = taosMemoryMalloc(nData);  // remove the copy (todo)
+  memcpy(pDataCopy, pData, nData);
+  tCoderInit(&coder, TD_LITTLE_ENDIAN, pDataCopy, nData, TD_DECODER);
+  ret = metaDecodeEntry(&coder, &me);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  type = me.type;
+  if (type == TSDB_CHILD_TABLE) {
+    ctime = me.ctbEntry.ctime;
+    suid = me.ctbEntry.suid;
+  } else if (type == TSDB_NORMAL_TABLE) {
+    ctime = me.ntbEntry.ctime;
+    suid = 0;
+  } else {
+    ASSERT(0);
+  }
+
+  taosMemoryFree(pDataCopy);
+  tCoderClear(&coder);
+  tdbDbcClose(pTbDbc);
+
+  if (type == TSDB_CHILD_TABLE) {
+    // remove the pCtbIdx
+    TDBC *pCtbIdxc = NULL;
+    tdbDbcOpen(pMeta->pCtbIdx, &pCtbIdxc, &pMeta->txn);
+
+    ret = tdbDbcMoveTo(pCtbIdxc, &(SCtbIdxKey){.suid = suid, .uid = uid}, sizeof(SCtbIdxKey), &c);
+    if (ret < 0 || c != 0) {
+      ASSERT(0);
+      return -1;
+    }
+
+    tdbDbcDelete(pCtbIdxc);
+    tdbDbcClose(pCtbIdxc);
+
+    // remove tags from pTagIdx (todo)
+  } else if (type == TSDB_NORMAL_TABLE) {
+    // remove from pSkmDb
+  } else {
+    ASSERT(0);
+  }
+
+  // remove from ttl (todo)
+  if (ctime > 0) {
+  }
 
   return 0;
 }
@@ -218,7 +339,7 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   tCoderClear(&coder);
 
   // write to table.db
-  if (tdbDbPut(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) {
+  if (tdbDbInsert(pMeta->pTbDb, pKey, kLen, pVal, vLen, &pMeta->txn) < 0) {
     goto _err;
   }
 
@@ -231,11 +352,11 @@ _err:
 }
 
 static int metaUpdateUidIdx(SMeta *pMeta, const SMetaEntry *pME) {
-  return tdbDbPut(pMeta->pUidIdx, &pME->uid, sizeof(tb_uid_t), &pME->version, sizeof(int64_t), &pMeta->txn);
+  return tdbDbInsert(pMeta->pUidIdx, &pME->uid, sizeof(tb_uid_t), &pME->version, sizeof(int64_t), &pMeta->txn);
 }
 
 static int metaUpdateNameIdx(SMeta *pMeta, const SMetaEntry *pME) {
-  return tdbDbPut(pMeta->pNameIdx, pME->name, strlen(pME->name) + 1, &pME->uid, sizeof(tb_uid_t), &pMeta->txn);
+  return tdbDbInsert(pMeta->pNameIdx, pME->name, strlen(pME->name) + 1, &pME->uid, sizeof(tb_uid_t), &pMeta->txn);
 }
 
 static int metaUpdateTtlIdx(SMeta *pMeta, const SMetaEntry *pME) {
@@ -258,12 +379,12 @@ static int metaUpdateTtlIdx(SMeta *pMeta, const SMetaEntry *pME) {
   ttlKey.dtime = ctime + ttlDays * 24 * 60 * 60;
   ttlKey.uid = pME->uid;
 
-  return tdbDbPut(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), NULL, 0, &pMeta->txn);
+  return tdbDbInsert(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), NULL, 0, &pMeta->txn);
 }
 
 static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) {
   SCtbIdxKey ctbIdxKey = {.suid = pME->ctbEntry.suid, .uid = pME->uid};
-  return tdbDbPut(pMeta->pCtbIdx, &ctbIdxKey, sizeof(ctbIdxKey), NULL, 0, &pMeta->txn);
+  return tdbDbInsert(pMeta->pCtbIdx, &ctbIdxKey, sizeof(ctbIdxKey), NULL, 0, &pMeta->txn);
 }
 
 static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pME) {
@@ -304,7 +425,7 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
   tCoderInit(&coder, TD_LITTLE_ENDIAN, pVal, vLen, TD_ENCODER);
   tEncodeSSchemaWrapper(&coder, pSW);
 
-  if (tdbDbPut(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) {
+  if (tdbDbInsert(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), pVal, vLen, &pMeta->txn) < 0) {
     rcode = -1;
     goto _exit;
   }

source/dnode/vnode/src/tsdb/tsdbTDBImpl.c

@@ -97,7 +97,7 @@ int32_t tsdbCloseDBF(SDBFile *pDBF) {
 int32_t tsdbSaveSmaToDB(SDBFile *pDBF, void *pKey, int32_t keyLen, void *pVal, int32_t valLen, TXN *txn) {
   int32_t ret;
 
-  ret = tdbDbPut(pDBF->pDB, pKey, keyLen, pVal, valLen, txn);
+  ret = tdbDbInsert(pDBF->pDB, pKey, keyLen, pVal, valLen, txn);
   if (ret < 0) {
     tsdbError("Failed to create insert sma data into db, ret = %d", ret);
     return -1;

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -445,14 +445,45 @@ static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM
 }
 
 static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
-  SVDropTbReq req = {0};
-  SVDropTbReq rsp = {0};
+  SVDropTbBatchReq req = {0};
+  SVDropTbBatchRsp rsp = {0};
+  SCoder           coder = {0};
+  int              ret;
+
+  pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
+  pRsp->pCont = NULL;
+  pRsp->contLen = 0;
+  pRsp->code = TSDB_CODE_SUCCESS;
 
   // decode req
+  tCoderInit(&coder, TD_LITTLE_ENDIAN, pReq, len, TD_DECODER);
+  ret = tDecodeSVDropTbBatchReq(&coder, &req);
+  if (ret < 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    pRsp->code = terrno;
+    goto _exit;
+  }
 
   // process req
+  rsp.pArray = taosArrayInit(sizeof(SVDropTbRsp), req.nReqs);
+  for (int iReq = 0; iReq < req.nReqs; iReq++) {
+    SVDropTbReq *pDropTbReq = req.pReqs + iReq;
+    SVDropTbRsp  dropTbRsp = {0};
 
-  // return rsp
+    /* code */
+    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq);
+    if (ret < 0) {
+      dropTbRsp.code = TSDB_CODE_SUCCESS;
+    } else {
+      dropTbRsp.code = terrno;
+    }
+
+    taosArrayPush(rsp.pArray, &dropTbRsp);
+  }
+
+_exit:
+  tCoderClear(&coder);
+  // encode rsp (TODO)
   return 0;
 }
 
@@ -482,7 +513,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 }
 
 int32_t tsdbProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
-  if(!pReq) {
+  if (!pReq) {
     terrno = TSDB_CODE_INVALID_PTR;
     return TSDB_CODE_FAILED;
   }

source/libs/tdb/inc/tdb.h

@@ -40,7 +40,9 @@ int tdbCommit(TENV *pEnv, TXN *pTxn);
 int tdbDbOpen(const char *fname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TENV *pEnv, TDB **ppDb);
 int tdbDbClose(TDB *pDb);
 int tdbDbDrop(TDB *pDb);
-int tdbDbPut(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn);
+int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn);
+int tdbDbDelete(TDB *pDb, const void *pKey, int kLen, TXN *pTxn);
+int tdbDbUpsert(TDB *pDb, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn);
 int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen);
 int tdbDbPGet(TDB *pDb, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);
 
@@ -53,11 +55,9 @@ int tdbDbcMoveToLast(TDBC *pDbc);
 int tdbDbcMoveToNext(TDBC *pDbc);
 int tdbDbcMoveToPrev(TDBC *pDbc);
 int tdbDbcGet(TDBC *pDbc, const void **ppKey, int *pkLen, const void **ppVal, int *pvLen);
-
-int tdbDbcPut(TDBC *pDbc, const void *pKey, int keyLen, const void *pVal, int valLen);
-int tdbDbcUpdate(TDBC *pDbc, const void *pKey, int kLen, const void *pVal, int vLen);
-int tdbDbcDrop(TDBC *pDbc);
+int tdbDbcDelete(TDBC *pDbc);
 int tdbDbcNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen);
+int tdbDbcUpsert(TDBC *pDbc, const void *pKey, int nKey, const void *pData, int nData, int insert);
 
 // TXN
 #define TDB_TXN_WRITE            0x1

source/libs/tdb/src/db/tdbBtree.c

@@ -138,67 +138,90 @@ int tdbBtreeInsert(SBTree *pBt, const void *pKey, int kLen, const void *pVal, in
   }
 
   if (btc.idx == -1) {
-    idx = 0;
+    btc.idx = 0;
   } else {
     if (c > 0) {
-      idx = btc.idx + 1;
-    } else if (c < 0) {
-      idx = btc.idx;
-    } else {
-      // TDB does NOT allow same key
-      tdbBtcClose(&btc);
+      btc.idx++;
+    } else if (c == 0) {
+      // dup key not allowed
       ASSERT(0);
       return -1;
     }
   }
 
-  // make sure enough space to hold the cell
-  szBuf = kLen + vLen + 14;
-  pBuf = tdbRealloc(pBt->pBuf, pBt->pageSize > szBuf ? szBuf : pBt->pageSize);
-  if (pBuf == NULL) {
-    tdbBtcClose(&btc);
+  ret = tdbBtcUpsert(&btc, pKey, kLen, pVal, vLen, 1);
+  if (ret < 0) {
     ASSERT(0);
+    tdbBtcClose(&btc);
     return -1;
   }
-  pBt->pBuf = pBuf;
-  pCell = (SCell *)pBt->pBuf;
 
-  // encode cell
-  ret = tdbBtreeEncodeCell(btc.pPage, pKey, kLen, pVal, vLen, pCell, &szCell);
+  tdbBtcClose(&btc);
+  return 0;
+}
+
+int tdbBtreeDelete(SBTree *pBt, const void *pKey, int kLen, TXN *pTxn) {
+  SBTC btc;
+  int  c;
+  int  ret;
+
+  tdbBtcOpen(&btc, pBt, pTxn);
+
+  // move the cursor
+  ret = tdbBtcMoveTo(&btc, pKey, kLen, &c);
   if (ret < 0) {
     tdbBtcClose(&btc);
     ASSERT(0);
     return -1;
   }
 
-  // mark the page dirty
-  ret = tdbPagerWrite(pBt->pPager, btc.pPage);
-  if (ret < 0) {
+  if (btc.idx < 0 || c != 0) {
     tdbBtcClose(&btc);
-    ASSERT(0);
     return -1;
   }
 
-  // insert the cell
-  ret = tdbPageInsertCell(btc.pPage, idx, pCell, szCell, 0);
-  if (ret < 0) {
+  // delete the key
+  if (tdbBtcDelete(&btc) < 0) {
     tdbBtcClose(&btc);
+    return -1;
+  }
+
+  tdbBtcClose(&btc);
+  return 0;
+}
+
+int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, int nData, TXN *pTxn) {
+  SBTC btc;
+  int  c;
+  int  ret;
+
+  tdbBtcOpen(&btc, pBt, pTxn);
+
+  // move the cursor
+  ret = tdbBtcMoveTo(&btc, pKey, nKey, &c);
+  if (ret < 0) {
     ASSERT(0);
+    tdbBtcClose(&btc);
     return -1;
   }
 
-  // check if need balance
-  if (btc.pPage->nOverflow > 0) {
-    ret = tdbBtreeBalance(&btc);
-    if (ret < 0) {
-      tdbBtcClose(&btc);
-      ASSERT(0);
-      return -1;
+  if (btc.idx == -1) {
+    btc.idx = 0;
+    c = 1;
+  } else {
+    if (c > 0) {
+      btc.idx = btc.idx + 1;
     }
   }
 
-  tdbBtcClose(&btc);
+  ret = tdbBtcUpsert(&btc, pKey, nKey, pData, nData, c);
+  if (ret < 0) {
+    ASSERT(0);
+    tdbBtcClose(&btc);
+    return -1;
+  }
 
+  tdbBtcClose(&btc);
   return 0;
 }
 
@@ -552,14 +575,14 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx, TXN *pTx
       SCell *pCell;
       int    szLCell, szRCell;
 
+      // balance page (iNew) and (iNew-1)
       for (;;) {
         pCell = tdbPageGetCell(pOlds[infoNews[iNew - 1].iPage], infoNews[iNew - 1].oIdx);
 
-        if (childNotLeaf) {
-          szLCell = szRCell = tdbBtreeCellSize(pOlds[infoNews[iNew - 1].iPage], pCell);
+        szLCell = tdbBtreeCellSize(pOlds[infoNews[iNew - 1].iPage], pCell);
+        if (!childNotLeaf) {
+          szRCell = szLCell;
         } else {
-          szLCell = tdbBtreeCellSize(pOlds[infoNews[iNew - 1].iPage], pCell);
-
           int    iPage = infoNews[iNew - 1].iPage;
           int    oIdx = infoNews[iNew - 1].oIdx + 1;
           SPage *pPage;
@@ -736,6 +759,13 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx, TXN *pTx
     }
   }
 
+  if (TDB_BTREE_PAGE_IS_ROOT(pParent) && TDB_PAGE_TOTAL_CELLS(pParent) == 0) {
+    i8 flags = TDB_BTREE_ROOT | TDB_BTREE_PAGE_IS_LEAF(pNews[0]);
+    // copy content to the parent page
+    tdbBtreeInitPage(pParent, &(SBtreeInitPageArg){.flags = flags, .pBt = pBt}, 0);
+    tdbPageCopy(pNews[0], pParent);
+  }
+
   for (int i = 0; i < 3; i++) {
     if (pDivCell[i]) {
       tdbOsFree(pDivCell[i]);
@@ -1357,7 +1387,143 @@ int tdbBtcGet(SBTC *pBtc, const void **ppKey, int *kLen, const void **ppVal, int
 
   if (ppVal) {
     *ppVal = (void *)pBtc->coder.pVal;
-    *kLen = pBtc->coder.vLen;
+    *vLen = pBtc->coder.vLen;
+  }
+
+  return 0;
+}
+
+int tdbBtcDelete(SBTC *pBtc) {
+  int         idx = pBtc->idx;
+  int         nCells = TDB_PAGE_TOTAL_CELLS(pBtc->pPage);
+  SPager     *pPager = pBtc->pBt->pPager;
+  const void *pKey;
+  i8          iPage;
+  SPage      *pPage;
+  SPgno       pgno;
+  SCell      *pCell;
+  int         szCell;
+  int         nKey;
+  int         ret;
+
+  ASSERT(idx >= 0 && idx < nCells);
+
+  // drop the cell on the leaf
+  ret = tdbPagerWrite(pPager, pBtc->pPage);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  tdbPageDropCell(pBtc->pPage, idx);
+
+  // update interior page or do balance
+  if (idx == nCells - 1) {
+    if (idx) {
+      pBtc->idx--;
+      tdbBtcGet(pBtc, &pKey, &nKey, NULL, NULL);
+
+      // loop to update the interial page
+      pgno = TDB_PAGE_PGNO(pBtc->pPage);
+      for (iPage = pBtc->iPage - 1; iPage >= 0; iPage--) {
+        pPage = pBtc->pgStack[iPage];
+        idx = pBtc->idxStack[iPage];
+        nCells = TDB_PAGE_TOTAL_CELLS(pPage);
+
+        if (idx < nCells) {
+          ret = tdbPagerWrite(pPager, pPage);
+          if (ret < 0) {
+            ASSERT(0);
+            return -1;
+          }
+
+          // update the cell with new key
+          pCell = tdbOsMalloc(nKey + 9);
+          tdbBtreeEncodeCell(pPage, pKey, nKey, &pgno, sizeof(pgno), pCell, &szCell);
+
+          ret = tdbPageUpdateCell(pPage, idx, pCell, szCell);
+          if (ret < 0) {
+            tdbOsFree(pCell);
+            ASSERT(0);
+            return -1;
+          }
+          tdbOsFree(pCell);
+          break;
+        } else {
+          pgno = TDB_PAGE_PGNO(pPage);
+        }
+      }
+    } else {
+      // delete the leaf page and do balance
+      ASSERT(TDB_PAGE_TOTAL_CELLS(pBtc->pPage) == 0);
+
+      ret = tdbBtreeBalance(pBtc);
+      if (ret < 0) {
+        ASSERT(0);
+        return -1;
+      }
+    }
+  }
+
+  return 0;
+}
+
+int tdbBtcUpsert(SBTC *pBtc, const void *pKey, int kLen, const void *pData, int nData, int insert) {
+  SCell *pCell;
+  int    szCell;
+  int    nCells = TDB_PAGE_TOTAL_CELLS(pBtc->pPage);
+  int    szBuf;
+  void  *pBuf;
+  int    ret;
+
+  ASSERT(pBtc->idx >= 0);
+
+  // alloc space
+  szBuf = kLen + nData + 14;
+  pBuf = tdbRealloc(pBtc->pBt->pBuf, pBtc->pBt->pageSize > szBuf ? szBuf : pBtc->pBt->pageSize);
+  if (pBuf == NULL) {
+    ASSERT(0);
+    return -1;
+  }
+  pBtc->pBt->pBuf = pBuf;
+  pCell = (SCell *)pBtc->pBt->pBuf;
+
+  // encode cell
+  ret = tdbBtreeEncodeCell(pBtc->pPage, pKey, kLen, pData, nData, pCell, &szCell);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  // mark dirty
+  ret = tdbPagerWrite(pBtc->pBt->pPager, pBtc->pPage);
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  // insert or update
+  if (insert) {
+    ASSERT(pBtc->idx <= nCells);
+
+    ret = tdbPageInsertCell(pBtc->pPage, pBtc->idx, pCell, szCell, 0);
+  } else {
+    ASSERT(pBtc->idx < nCells);
+
+    ret = tdbPageUpdateCell(pBtc->pPage, pBtc->idx, pCell, szCell);
+  }
+  if (ret < 0) {
+    ASSERT(0);
+    return -1;
+  }
+
+  // check balance
+  if (pBtc->pPage->nOverflow > 0) {
+    ret = tdbBtreeBalance(pBtc);
+    if (ret < 0) {
+      ASSERT(0);
+      return -1;
+    }
   }
 
   return 0;

source/libs/tdb/src/db/tdbDb.c

@@ -75,10 +75,16 @@ int tdbDbDrop(TDB *pDb) {
   return 0;
 }
 
-int tdbDbPut(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn) {
+int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn) {
   return tdbBtreeInsert(pDb->pBt, pKey, keyLen, pVal, valLen, pTxn);
 }
 
+int tdbDbDelete(TDB *pDb, const void *pKey, int kLen, TXN *pTxn) { return tdbBtreeDelete(pDb->pBt, pKey, kLen, pTxn); }
+
+int tdbDbUpsert(TDB *pDb, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn) {
+  return tdbBtreeUpsert(pDb->pBt, pKey, kLen, pVal, vLen, pTxn);
+}
+
 int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen) {
   return tdbBtreeGet(pDb->pBt, pKey, kLen, ppVal, vLen);
 }
@@ -117,28 +123,16 @@ int tdbDbcGet(TDBC *pDbc, const void **ppKey, int *pkLen, const void **ppVal, in
   return tdbBtcGet(&pDbc->btc, ppKey, pkLen, ppVal, pvLen);
 }
 
-int tdbDbcPut(TDBC *pDbc, const void *pKey, int keyLen, const void *pVal, int valLen) {
-  // TODO
-  ASSERT(0);
-  return 0;
-}
-
-int tdbDbcUpdate(TDBC *pDbc, const void *pKey, int kLen, const void *pVal, int vLen) {
-  // TODO
-  ASSERT(0);
-  return 0;
-}
-
-int tdbDbcDrop(TDBC *pDbc) {
-  // TODO
-  ASSERT(0);
-  return 0;
-}
+int tdbDbcDelete(TDBC *pDbc) { return tdbBtcDelete(&pDbc->btc); }
 
 int tdbDbcNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen) {
   return tdbBtreeNext(&pDbc->btc, ppKey, kLen, ppVal, vLen);
 }
 
+int tdbDbcUpsert(TDBC *pDbc, const void *pKey, int nKey, const void *pData, int nData, int insert) {
+  return tdbBtcUpsert(&pDbc->btc, pKey, nKey, pData, nData, insert);
+}
+
 int tdbDbcClose(TDBC *pDbc) {
   if (pDbc) {
     tdbBtcClose(&pDbc->btc);

source/libs/tdb/src/db/tdbPage.c

@@ -171,6 +171,11 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl
   return 0;
 }
 
+int tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
+  tdbPageDropCell(pPage, idx);
+  return tdbPageInsertCell(pPage, idx, pCell, szCell, 0);
+}
+
 int tdbPageDropCell(SPage *pPage, int idx) {
   int    lidx;
   SCell *pCell;

source/libs/tdb/src/inc/tdbInt.h

@@ -128,6 +128,8 @@ struct SBTC {
 int tdbBtreeOpen(int keyLen, int valLen, SPager *pFile, tdb_cmpr_fn_t kcmpr, SBTree **ppBt);
 int tdbBtreeClose(SBTree *pBt);
 int tdbBtreeInsert(SBTree *pBt, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn);
+int tdbBtreeDelete(SBTree *pBt, const void *pKey, int kLen, TXN *pTxn);
+int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, int nData, TXN *pTxn);
 int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen);
 int tdbBtreePGet(SBTree *pBt, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen);
 
@@ -141,6 +143,8 @@ int tdbBtcMoveToNext(SBTC *pBtc);
 int tdbBtcMoveToPrev(SBTC *pBtc);
 int tdbBtreeNext(SBTC *pBtc, void **ppKey, int *kLen, void **ppVal, int *vLen);
 int tdbBtcGet(SBTC *pBtc, const void **ppKey, int *kLen, const void **ppVal, int *vLen);
+int tdbBtcDelete(SBTC *pBtc);
+int tdbBtcUpsert(SBTC *pBtc, const void *pKey, int kLen, const void *pData, int nData, int insert);
 
 // tdbPager.c ====================================
 
@@ -278,6 +282,7 @@ void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell
 void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *));
 int  tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl);
 int  tdbPageDropCell(SPage *pPage, int idx);
+int  tdbPageUpdateCell(SPage *pPage, int idx, SCell *pCell, int szCell);
 void tdbPageCopy(SPage *pFromPage, SPage *pToPage);
 int  tdbPageCapacity(int pageSize, int amHdrSize);
 

source/libs/tdb/test/tdbTest.cpp

@@ -115,12 +115,12 @@ static int tDefaultKeyCmpr(const void *pKey1, int keyLen1, const void *pKey2, in
   return cret;
 }
 
-TEST(tdb_test, simple_test) {
+TEST(tdb_test, simple_insert1) {
   int           ret;
   TENV         *pEnv;
   TDB          *pDb;
   tdb_cmpr_fn_t compFunc;
-  int           nData = 10000000;
+  int           nData = 1000000;
   TXN           txn;
 
   taosRemoveDir("tdb");
@@ -152,7 +152,7 @@ TEST(tdb_test, simple_test) {
     for (int iData = 1; iData <= nData; iData++) {
       sprintf(key, "key%d", iData);
       sprintf(val, "value%d", iData);
-      ret = tdbDbPut(pDb, key, strlen(key), val, strlen(val), &txn);
+      ret = tdbDbInsert(pDb, key, strlen(key), val, strlen(val), &txn);
       GTEST_ASSERT_EQ(ret, 0);
 
       // if pool is full, commit the transaction and start a new one
@@ -202,6 +202,8 @@ TEST(tdb_test, simple_test) {
       ret = tdbDbcOpen(pDb, &pDBC, NULL);
       GTEST_ASSERT_EQ(ret, 0);
 
+      tdbDbcMoveToFirst(pDBC);
+
       for (;;) {
         ret = tdbDbcNext(pDBC, &pKey, &kLen, &pVal, &vLen);
         if (ret < 0) break;
@@ -233,7 +235,7 @@ TEST(tdb_test, simple_test) {
   GTEST_ASSERT_EQ(ret, 0);
 }
 
-TEST(tdb_test, simple_test2) {
+TEST(tdb_test, simple_insert2) {
   int           ret;
   TENV         *pEnv;
   TDB          *pDb;
@@ -269,7 +271,7 @@ TEST(tdb_test, simple_test2) {
     for (int iData = 1; iData <= nData; iData++) {
       sprintf(key, "key%d", iData);
       sprintf(val, "value%d", iData);
-      ret = tdbDbPut(pDb, key, strlen(key), val, strlen(val), &txn);
+      ret = tdbDbInsert(pDb, key, strlen(key), val, strlen(val), &txn);
       GTEST_ASSERT_EQ(ret, 0);
     }
 
@@ -283,13 +285,15 @@ TEST(tdb_test, simple_test2) {
       ret = tdbDbcOpen(pDb, &pDBC, NULL);
       GTEST_ASSERT_EQ(ret, 0);
 
+      tdbDbcMoveToFirst(pDBC);
+
       for (;;) {
         ret = tdbDbcNext(pDBC, &pKey, &kLen, &pVal, &vLen);
         if (ret < 0) break;
 
-        std::cout.write((char *)pKey, kLen) /* << " " << kLen */ << " ";
-        std::cout.write((char *)pVal, vLen) /* << " " << vLen */;
-        std::cout << std::endl;
+        // std::cout.write((char *)pKey, kLen) /* << " " << kLen */ << " ";
+        // std::cout.write((char *)pVal, vLen) /* << " " << vLen */;
+        // std::cout << std::endl;
 
         count++;
       }
@@ -316,4 +320,164 @@ TEST(tdb_test, simple_test2) {
   // Close Env
   ret = tdbEnvClose(pEnv);
   GTEST_ASSERT_EQ(ret, 0);
+}
+
+TEST(tdb_test, simple_delete1) {
+  int       ret;
+  TDB      *pDb;
+  char      key[128];
+  char      data[128];
+  TXN       txn;
+  TENV     *pEnv;
+  SPoolMem *pPool;
+  void     *pKey = NULL;
+  void     *pData = NULL;
+  int       nKey;
+  TDBC     *pDbc;
+  int       nData;
+  int       nKV = 69;
+
+  taosRemoveDir("tdb");
+
+  pPool = openPool();
+
+  // open env
+  ret = tdbEnvOpen("tdb", 1024, 256, &pEnv);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // open database
+  ret = tdbDbOpen("db.db", -1, -1, tKeyCmpr, pEnv, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  tdbTxnOpen(&txn, 0, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+  tdbBegin(pEnv, &txn);
+
+  // loop to insert batch data
+  for (int iData = 0; iData < nKV; iData++) {
+    sprintf(key, "key%d", iData);
+    sprintf(data, "data%d", iData);
+    ret = tdbDbInsert(pDb, key, strlen(key), data, strlen(data), &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  // query the data
+  for (int iData = 0; iData < nKV; iData++) {
+    sprintf(key, "key%d", iData);
+    sprintf(data, "data%d", iData);
+
+    ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
+    GTEST_ASSERT_EQ(ret, 0);
+    GTEST_ASSERT_EQ(memcmp(data, pData, nData), 0);
+  }
+
+  // loop to delete some data
+  for (int iData = nKV - 1; iData > 30; iData--) {
+    sprintf(key, "key%d", iData);
+
+    ret = tdbDbDelete(pDb, key, strlen(key), &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  // query the data
+  for (int iData = 0; iData < nKV; iData++) {
+    sprintf(key, "key%d", iData);
+
+    ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
+    if (iData <= 30) {
+      GTEST_ASSERT_EQ(ret, 0);
+    } else {
+      GTEST_ASSERT_EQ(ret, -1);
+    }
+  }
+
+  // loop to iterate the data
+  tdbDbcOpen(pDb, &pDbc, NULL);
+
+  ret = tdbDbcMoveToFirst(pDbc);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  pKey = NULL;
+  pData = NULL;
+  for (;;) {
+    ret = tdbDbcNext(pDbc, &pKey, &nKey, &pData, &nData);
+    if (ret < 0) break;
+
+    std::cout.write((char *)pKey, nKey) /* << " " << kLen */ << " ";
+    std::cout.write((char *)pData, nData) /* << " " << vLen */;
+    std::cout << std::endl;
+  }
+
+  tdbDbcClose(pDbc);
+
+  tdbCommit(pEnv, &txn);
+
+  closePool(pPool);
+
+  tdbDbClose(pDb);
+  tdbEnvClose(pEnv);
+}
+
+TEST(tdb_test, simple_upsert1) {
+  int       ret;
+  TENV     *pEnv;
+  TDB      *pDb;
+  int       nData = 100000;
+  char      key[64];
+  char      data[64];
+  void     *pData = NULL;
+  SPoolMem *pPool;
+  TXN       txn;
+
+  taosRemoveDir("tdb");
+
+  // open env
+  ret = tdbEnvOpen("tdb", 4096, 64, &pEnv);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  // open database
+  ret = tdbDbOpen("db.db", -1, -1, NULL, pEnv, &pDb);
+  GTEST_ASSERT_EQ(ret, 0);
+
+  pPool = openPool();
+  // insert some data
+  tdbTxnOpen(&txn, 0, poolMalloc, poolFree, pPool, TDB_TXN_WRITE | TDB_TXN_READ_UNCOMMITTED);
+  tdbBegin(pEnv, &txn);
+
+  for (int iData = 0; iData < nData; iData++) {
+    sprintf(key, "key%d", iData);
+    sprintf(data, "data%d", iData);
+    ret = tdbDbInsert(pDb, key, strlen(key), data, strlen(data), &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  // query the data
+  for (int iData = 0; iData < nData; iData++) {
+    sprintf(key, "key%d", iData);
+    sprintf(data, "data%d", iData);
+    ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
+    GTEST_ASSERT_EQ(ret, 0);
+    GTEST_ASSERT_EQ(memcmp(pData, data, nData), 0);
+  }
+
+  // upsert some data
+  for (int iData = 0; iData < nData; iData++) {
+    sprintf(key, "key%d", iData);
+    sprintf(data, "data%d-u", iData);
+    ret = tdbDbUpsert(pDb, key, strlen(key), data, strlen(data), &txn);
+    GTEST_ASSERT_EQ(ret, 0);
+  }
+
+  tdbCommit(pEnv, &txn);
+
+  // query the data
+  for (int iData = 0; iData < nData; iData++) {
+    sprintf(key, "key%d", iData);
+    sprintf(data, "data%d-u", iData);
+    ret = tdbDbGet(pDb, key, strlen(key), &pData, &nData);
+    GTEST_ASSERT_EQ(ret, 0);
+    GTEST_ASSERT_EQ(memcmp(pData, data, nData), 0);
+  }
+
+  tdbDbClose(pDb);
+  tdbEnvClose(pEnv);
 }
\ No newline at end of file

source/util/src/tskiplist2.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "tskiplist2.h"
+
+struct SSLNode {
+  int8_t   level;
+  SSLNode *forwards[];
+};
+
+struct SSkipList2 {
+  int8_t        level;
+  uint32_t      seed;
+  int32_t       size;
+  const SSLCfg *pCfg;
+  SSLNode      *pHead[];
+};
+
+static void   *slMalloc(void *pPool, int32_t size);
+static void    slFree(void *pPool, void *p);
+static int32_t slCmprFn(const void *pKey, int32_t nKey, const void *pData, int32_t nData);
+
+const SSLCfg slDefaultCfg = {.maxLevel = SL_MAX_LEVEL,
+                             .nKey = -1,
+                             .nData = -1,
+                             .cmprFn = slCmprFn,
+                             .pPool = NULL,
+                             .xMalloc = slMalloc,
+                             .xFree = slFree};
+
+int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl) {
+  SSkipList2 *pSl = NULL;
+  int32_t     size;
+
+  *ppSl = NULL;
+  if (pCfg == NULL) pCfg = &slDefaultCfg;
+
+  // check config (TODO)
+
+  // malloc handle
+  size = sizeof(*pSl) + sizeof(SSLNode *) * pCfg->maxLevel * 2;
+  pSl = pCfg->xMalloc(pCfg->pPool, size);
+  if (pSl == NULL) {
+    return -1;
+  }
+
+  pSl->level = 0;
+  pSl->seed = taosRand();
+  pSl->size = 0;
+  pSl->pCfg = pCfg;
+
+  // init an empty skiplist
+  for (int32_t i = 0; i < pCfg->maxLevel * 2; i++) {
+    pSl->pHead[i] = NULL;
+  }
+
+  *ppSl = pSl;
+  return 0;
+}
+
+int32_t slClose(SSkipList2 *pSl) {
+  if (pSl) {
+    slClear(pSl);
+    if (pSl->pCfg->xFree) {
+      pSl->pCfg->xFree(pSl->pCfg->pPool, pSl);
+    }
+  }
+
+  return 0;
+}
+
+int32_t slClear(SSkipList2 *pSl) {
+  // loop to clear sl
+  for (;;) {
+    // (TODO)
+  }
+
+  // init sl (TODO)
+
+  return 0;
+}
+
+int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc) {
+  pSlc->pSl = pSl;
+
+  for (int i = 0; i < SL_MAX_LEVEL; i++) {
+    if (i < pSl->pCfg->maxLevel) {
+    } else {
+      pSlc->forwards[i] = NULL;
+    }
+  }
+
+  // TODO
+  return 0;
+}
+
+int32_t slcClose(SSLCursor *pSlc) {
+  // TODO
+  return 0;
+}
+
+int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey) {
+  // TODO
+  return 0;
+}
+
+int32_t slcMoveToNext(SSLCursor *pSlc) {
+  // TODO
+  return 0;
+}
+
+int32_t slcMoveToPrev(SSLCursor *pSlc) {
+  // TODO
+  return 0;
+}
+
+int32_t slcMoveToFirst(SSLCursor *pSlc) {
+  // TODO
+  return 0;
+}
+
+int32_t slcMoveToLast(SSLCursor *pSlc) {
+  // TODO
+  return 0;
+}
+
+int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData) {
+  // TODO
+  return 0;
+}
+
+int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData) {
+  // TODO
+  return 0;
+}
+
+int32_t slcDrop(SSLCursor *pSlc) {
+  // TODO
+  return 0;
+}
+
+static FORCE_INLINE void *slMalloc(void *pPool, int32_t size) { return taosMemoryMalloc(size); }
+
+static FORCE_INLINE void slFree(void *pPool, void *p) { taosMemoryFree(p); }
+
+static int32_t slCmprFn(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2) {
+  ASSERT(nKey1 >= 0 && nKey2 >= 0);
+
+  int32_t nKey = nKey1 > nKey2 ? nKey2 : nKey1;
+  int32_t c;
+
+  c = memcmp(pKey1, pKey2, nKey);
+  if (c == 0) {
+    if (nKey1 > nKey2) {
+      c = 1;
+    } else if (nKey1 < nKey2) {
+      c = -1;
+    }
+  }
+
+  return c;
+}
\ No newline at end of file