Merge pull request #4239 from taosdata/feature/TD-2088

Feature/td 2088

src/tsdb/inc/tsdbMain.h

@@ -208,6 +208,18 @@ typedef struct {
 } SFileGroupIter;
 
 // ------------------ tsdbMain.c
+typedef struct {
+  int32_t  totalLen;
+  int32_t  len;
+  SDataRow row;
+} SSubmitBlkIter;
+
+typedef struct {
+  int32_t totalLen;
+  int32_t len;
+  void *  pMsg;
+} SSubmitMsgIter;
+
 typedef struct {
   int8_t state;
 
@@ -430,7 +442,6 @@ void          tsdbCloseBufPool(STsdbRepo* pRepo);
 SListNode*    tsdbAllocBufBlockFromPool(STsdbRepo* pRepo);
 
 // ------------------ tsdbMemTable.c
-int   tsdbUpdateRowInMem(STsdbRepo* pRepo, SDataRow row, STable* pTable);
 int   tsdbRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem);

src/tsdb/src/tsdbFile.c

@@ -516,7 +516,7 @@ void tsdbGetFileInfoImpl(char *fname, uint32_t *magic, int64_t *size) {
   SFile         file;
   SFile *       pFile = &file;
 
-  strncpy(pFile->fname, fname, TSDB_FILENAME_LEN);
+  strncpy(pFile->fname, fname, TSDB_FILENAME_LEN - 1);
   pFile->fd = -1;
 
   if (tsdbOpenFile(pFile, O_RDONLY) < 0) goto _err;

src/tsdb/src/tsdbMain.c

@@ -32,18 +32,6 @@
 #define TSDB_DEFAULT_COMPRESSION TWO_STAGE_COMP
 #define IS_VALID_COMPRESSION(compression) (((compression) >= NO_COMPRESSION) && ((compression) <= TWO_STAGE_COMP))
 
-typedef struct {
-  int32_t  totalLen;
-  int32_t  len;
-  SDataRow row;
-} SSubmitBlkIter;
-
-typedef struct {
-  int32_t totalLen;
-  int32_t len;
-  void *  pMsg;
-} SSubmitMsgIter;
-
 static int32_t     tsdbCheckAndSetDefaultCfg(STsdbCfg *pCfg);
 static int32_t     tsdbSetRepoEnv(char *rootDir, STsdbCfg *pCfg);
 static int32_t     tsdbUnsetRepoEnv(char *rootDir);
@@ -52,20 +40,13 @@ static int         tsdbLoadConfig(char *rootDir, STsdbCfg *pCfg);
 static char *      tsdbGetCfgFname(char *rootDir);
 static STsdbRepo * tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg);
 static void        tsdbFreeRepo(STsdbRepo *pRepo);
-static int         tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter);
-static int32_t     tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, TSKEY now, int32_t *affectedrows);
-static int         tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock);
-static SDataRow    tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter);
 static int         tsdbRestoreInfo(STsdbRepo *pRepo);
-static int         tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter);
 static void        tsdbAlterCompression(STsdbRepo *pRepo, int8_t compression);
 static int         tsdbAlterKeep(STsdbRepo *pRepo, int32_t keep);
 static int         tsdbAlterCacheTotalBlocks(STsdbRepo *pRepo, int totalBlocks);
 static int         keyFGroupCompFunc(const void *key, const void *fgroup);
 static int         tsdbEncodeCfg(void **buf, STsdbCfg *pCfg);
 static void *      tsdbDecodeCfg(void *buf, STsdbCfg *pCfg);
-static int         tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable);
-static int         tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg);
 static void        tsdbStartStream(STsdbRepo *pRepo);
 static void        tsdbStopStream(STsdbRepo *pRepo);
 
@@ -177,40 +158,6 @@ void tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit) {
   tsdbDebug("vgId:%d repository is closed", vgId);
 }
 
-int32_t tsdbInsertData(TSDB_REPO_T *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pRsp) {
-  STsdbRepo *    pRepo = (STsdbRepo *)repo;
-  SSubmitMsgIter msgIter = {0};
-
-  if (tsdbScanAndConvertSubmitMsg(pRepo, pMsg) < 0) {
-    if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-      tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pRepo), tstrerror(terrno));
-    }
-    return -1;
-  }
-
-  if (tsdbInitSubmitMsgIter(pMsg, &msgIter) < 0) {
-    tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pRepo), tstrerror(terrno));
-    return -1;
-  }
-
-  SSubmitBlk *pBlock = NULL;
-  int32_t     affectedrows = 0;
-
-  TSKEY now = taosGetTimestamp(pRepo->config.precision);
-  while (true) {
-    tsdbGetSubmitMsgNext(&msgIter, &pBlock);
-    if (pBlock == NULL) break;
-    if (tsdbInsertDataToTable(pRepo, pBlock, now, &affectedrows) < 0) {
-      return -1;
-    }
-  }
-
-  if (pRsp != NULL) pRsp->affectedRows = htonl(affectedrows);
-
-  if (tsdbCheckCommit(pRepo) < 0) return -1;
-  return 0;
-}
-
 uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_t eindex, int64_t *size) {
   STsdbRepo *pRepo = (STsdbRepo *)repo;
   // STsdbMeta *pMeta = pRepo->tsdbMeta;
@@ -735,93 +682,6 @@ static void tsdbFreeRepo(STsdbRepo *pRepo) {
   }
 }
 
-static int tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter) {
-  if (pMsg == NULL) {
-    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
-    return -1;
-  }
-
-  pIter->totalLen = pMsg->length;
-  pIter->len = 0;
-  pIter->pMsg = pMsg;
-  if (pMsg->length <= TSDB_SUBMIT_MSG_HEAD_SIZE) {
-    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
-    return -1;
-  }
-
-  return 0;
-}
-
-static int32_t tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, TSKEY now, int32_t *affectedrows) {
-  STsdbMeta *pMeta = pRepo->tsdbMeta;
-  int64_t    points = 0;
-
-  ASSERT(pBlock->tid < pMeta->maxTables);
-  STable *pTable = pMeta->tables[pBlock->tid];
-  ASSERT(pTable != NULL && TABLE_UID(pTable) == pBlock->uid);
-
-  SSubmitBlkIter blkIter = {0};
-  SDataRow       row = NULL;
-
-  TSKEY minKey = now - tsMsPerDay[pRepo->config.precision] * pRepo->config.keep;
-  TSKEY maxKey = now + tsMsPerDay[pRepo->config.precision] * pRepo->config.daysPerFile;
-
-  tsdbInitSubmitBlkIter(pBlock, &blkIter);
-  while ((row = tsdbGetSubmitBlkNext(&blkIter)) != NULL) {
-    if (dataRowKey(row) < minKey || dataRowKey(row) > maxKey) {
-      tsdbError("vgId:%d table %s tid %d uid %" PRIu64 " timestamp is out of range! now %" PRId64 " minKey %" PRId64
-                " maxKey %" PRId64,
-                REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), now, minKey, maxKey);
-      terrno = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
-      return -1;
-    }
-
-    if (tsdbUpdateRowInMem(pRepo, row, pTable) < 0) return -1;
-
-    (*affectedrows)++;
-    points++;
-  }
-
-  STSchema *pSchema = tsdbGetTableSchemaByVersion(pTable, pBlock->sversion);
-  pRepo->stat.pointsWritten += points * schemaNCols(pSchema);
-  pRepo->stat.totalStorage += points * schemaVLen(pSchema);
-
-  return 0;
-}
-
-static int tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
-  if (pIter->len == 0) {
-    pIter->len += TSDB_SUBMIT_MSG_HEAD_SIZE;
-  } else {
-    SSubmitBlk *pSubmitBlk = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
-    pIter->len += (sizeof(SSubmitBlk) + pSubmitBlk->dataLen + pSubmitBlk->schemaLen);
-  }
-
-  if (pIter->len > pIter->totalLen) {
-    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
-    *pPBlock = NULL;
-    return -1;
-  }
-
-  *pPBlock = (pIter->len == pIter->totalLen) ? NULL : (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
-
-  return 0;
-}
-
-static SDataRow tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter) {
-  SDataRow row = pIter->row;
-  if (row == NULL) return NULL;
-
-  pIter->len += dataRowLen(row);
-  if (pIter->len >= pIter->totalLen) {
-    pIter->row = NULL;
-  } else {
-    pIter->row = (char *)row + dataRowLen(row);
-  }
-
-  return row;
-}
-
 static int tsdbRestoreInfo(STsdbRepo *pRepo) {
   STsdbMeta * pMeta = pRepo->tsdbMeta;
   STsdbFileH *pFileH = pRepo->tsdbFileH;
@@ -855,14 +715,6 @@ _err:
   return -1;
 }
 
-static int tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
-  if (pBlock->dataLen <= 0) return -1;
-  pIter->totalLen = pBlock->dataLen;
-  pIter->len = 0;
-  pIter->row = (SDataRow)(pBlock->data+pBlock->schemaLen);
-  return 0;
-}
-
 static void tsdbAlterCompression(STsdbRepo *pRepo, int8_t compression) {
   int8_t ocompression = pRepo->config.compression;
   pRepo->config.compression = compression;
@@ -959,134 +811,6 @@ static void *tsdbDecodeCfg(void *buf, STsdbCfg *pCfg) {
   return buf;
 }
 
-static int tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable) {
-  ASSERT(pTable != NULL);
-
-  STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
-  int       sversion = schemaVersion(pSchema);
-
-  if (pBlock->sversion == sversion) {
-    return 0;
-  } else {
-    if (TABLE_TYPE(pTable) == TSDB_STREAM_TABLE) {  // stream table is not allowed to change schema
-      terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
-      return -1;
-    }
-  }
-
-  if (pBlock->sversion > sversion) {  // may need to update table schema
-    if (pBlock->schemaLen > 0) {
-      tsdbDebug(
-          "vgId:%d table %s tid %d uid %" PRIu64 " schema version %d is out of data, client version %d, update...",
-          REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), sversion, pBlock->sversion);
-      ASSERT(pBlock->schemaLen % sizeof(STColumn) == 0);
-      int       numOfCols = pBlock->schemaLen / sizeof(STColumn);
-      STColumn *pTCol = (STColumn *)pBlock->data;
-
-      STSchemaBuilder schemaBuilder = {0};
-      if (tdInitTSchemaBuilder(&schemaBuilder, pBlock->sversion) < 0) {
-        terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-        tsdbError("vgId:%d failed to update schema of table %s since %s", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable),
-                  tstrerror(terrno));
-        return -1;
-      }
-
-      for (int i = 0; i < numOfCols; i++) {
-        if (tdAddColToSchema(&schemaBuilder, pTCol[i].type, htons(pTCol[i].colId), htons(pTCol[i].bytes)) < 0) {
-          terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-          tsdbError("vgId:%d failed to update schema of table %s since %s", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable),
-                    tstrerror(terrno));
-          tdDestroyTSchemaBuilder(&schemaBuilder);
-          return -1;
-        }
-      }
-
-      STSchema *pNSchema = tdGetSchemaFromBuilder(&schemaBuilder);
-      if (pNSchema == NULL) {
-        terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-        tdDestroyTSchemaBuilder(&schemaBuilder);
-        return -1;
-      }
-
-      tdDestroyTSchemaBuilder(&schemaBuilder);
-      tsdbUpdateTableSchema(pRepo, pTable, pNSchema, true);
-    } else {
-      tsdbDebug(
-          "vgId:%d table %s tid %d uid %" PRIu64 " schema version %d is out of data, client version %d, reconfigure...",
-          REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), sversion, pBlock->sversion);
-      terrno = TSDB_CODE_TDB_TABLE_RECONFIGURE;
-      return -1;
-    }
-  } else {
-    ASSERT(pBlock->sversion >= 0);
-    if (tsdbGetTableSchemaImpl(pTable, false, false, pBlock->sversion) == NULL) {
-      tsdbError("vgId:%d invalid submit schema version %d to table %s tid %d from client", REPO_ID(pRepo),
-                pBlock->sversion, TABLE_CHAR_NAME(pTable), TABLE_TID(pTable));
-    }
-    terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
-    return -1;
-  }
-
-  return 0;
-}
-
-static int tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg) {
-  ASSERT(pMsg != NULL);
-  STsdbMeta *    pMeta = pRepo->tsdbMeta;
-  SSubmitMsgIter msgIter = {0};
-  SSubmitBlk *   pBlock = NULL;
-
-  terrno = TSDB_CODE_SUCCESS;
-  pMsg->length = htonl(pMsg->length);
-  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
-
-  if (tsdbInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
-  while (true) {
-    if (tsdbGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
-    if (pBlock == NULL) break;
-
-    pBlock->uid = htobe64(pBlock->uid);
-    pBlock->tid = htonl(pBlock->tid);
-    pBlock->sversion = htonl(pBlock->sversion);
-    pBlock->dataLen = htonl(pBlock->dataLen);
-    pBlock->schemaLen = htonl(pBlock->schemaLen);
-    pBlock->numOfRows = htons(pBlock->numOfRows);
-
-    if (pBlock->tid <= 0 || pBlock->tid >= pMeta->maxTables) {
-      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pRepo), pBlock->uid,
-                pBlock->tid);
-      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-      return -1;
-    }
-
-    STable *pTable = pMeta->tables[pBlock->tid];
-    if (pTable == NULL || TABLE_UID(pTable) != pBlock->uid) {
-      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pRepo), pBlock->uid,
-                pBlock->tid);
-      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-      return -1;
-    }
-
-    if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
-      tsdbError("vgId:%d invalid action trying to insert a super table %s", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable));
-      terrno = TSDB_CODE_TDB_INVALID_ACTION;
-      return -1;
-    }
-
-    // Check schema version and update schema if needed
-    if (tsdbCheckTableSchema(pRepo, pBlock, pTable) < 0) {
-      if (terrno == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-        continue;
-      } else {
-        return -1;
-      }
-    }
-  }
-
-  if (terrno != TSDB_CODE_SUCCESS) return -1;
-  return 0;
-}
-
 static int tsdbAlterCacheTotalBlocks(STsdbRepo *pRepo, int totalBlocks) {
   // TODO
   // STsdbCache *pCache = pRepo->tsdbCache;

src/tsdb/src/tsdbRWHelper.c

@@ -1595,7 +1595,7 @@ static int tsdbProcessMergeCommit(SRWHelper *pHelper, SCommitIter *pCommitIter,
       tblkIdx++;
     } else if (oBlock.numOfRows + pMergeInfo->rowsInserted - pMergeInfo->rowsDeleteSucceed == 0) {
       // Delete the block and do some stuff
-      ASSERT(pMergeInfo->keyFirst == INT64_MAX && pMergeInfo->keyFirst == INT64_MIN);
+      // ASSERT(pMergeInfo->keyFirst == INT64_MAX && pMergeInfo->keyFirst == INT64_MIN);
       if (tsdbDeleteSuperBlock(pHelper, tblkIdx) < 0) return -1;
       *pCommitIter->pIter = slIter;
       if (oBlock.last && pHelper->hasOldLastBlock) pHelper->hasOldLastBlock = false;

src/util/inc/tskiplist.h

@@ -131,6 +131,7 @@ SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint16_t keyLen, _
                            __sl_key_fn_t fn);
 void       tSkipListDestroy(SSkipList *pSkipList);
 SSkipListNode *    tSkipListPut(SSkipList *pSkipList, void *pData);
+void               tSkipListPutBatch(SSkipList *pSkipList, void **ppData, int ndata);
 SArray *           tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey);
 void               tSkipListPrint(SSkipList *pSkipList, int16_t nlevel);
 SSkipListIterator *tSkipListCreateIter(SSkipList *pSkipList);

src/util/src/tskiplist.c

@@ -24,10 +24,12 @@ static SSkipListNode *    getPriorNode(SSkipList *pSkipList, const char *val, in
 static void               tSkipListRemoveNodeImpl(SSkipList *pSkipList, SSkipListNode *pNode);
 static void               tSkipListCorrectLevel(SSkipList *pSkipList);
 static SSkipListIterator *doCreateSkipListIterator(SSkipList *pSkipList, int32_t order);
-static void               tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **backward, SSkipListNode *pNode);
-static bool               tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward, void *pData);
-static SSkipListNode *    tSkipListNewNode(uint8_t level);
+static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **direction, SSkipListNode *pNode, bool isForward);
+static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward, void *pData);
+static SSkipListNode *tSkipListNewNode(uint8_t level);
 #define tSkipListFreeNode(n) tfree((n))
+static SSkipListNode *tSkipListPutImpl(SSkipList *pSkipList, void *pData, SSkipListNode **direction, bool isForward,
+                                       bool hasDup);
 
 static FORCE_INLINE int     tSkipListWLock(SSkipList *pSkipList);
 static FORCE_INLINE int     tSkipListRLock(SSkipList *pSkipList);
@@ -109,30 +111,85 @@ SSkipListNode *tSkipListPut(SSkipList *pSkipList, void *pData) {
   if (pSkipList == NULL || pData == NULL) return NULL;
 
   SSkipListNode *backward[MAX_SKIP_LIST_LEVEL] = {0};
-  uint8_t        dupMode = SL_DUP_MODE(pSkipList);
   SSkipListNode *pNode = NULL;
 
   tSkipListWLock(pSkipList);
 
   bool hasDup = tSkipListGetPosToPut(pSkipList, backward, pData);
+  pNode = tSkipListPutImpl(pSkipList, pData, backward, false, hasDup);
 
-  if (hasDup && (dupMode == SL_DISCARD_DUP_KEY || dupMode == SL_UPDATE_DUP_KEY)) {
-    if (dupMode == SL_UPDATE_DUP_KEY) {
-      pNode = SL_NODE_GET_BACKWARD_POINTER(backward[0], 0);
-      atomic_store_ptr(&(pNode->pData), pData);
-    }
-  } else {
-    pNode = tSkipListNewNode(getSkipListRandLevel(pSkipList));
-    if (pNode != NULL) {
-      pNode->pData = pData;
+  tSkipListUnlock(pSkipList);
+
+  return pNode;
+}
+
+// Put a batch of data into skiplist. The batch of data must be in ascending order
+void tSkipListPutBatch(SSkipList *pSkipList, void **ppData, int ndata) {
+  SSkipListNode *backward[MAX_SKIP_LIST_LEVEL] = {0};
+  SSkipListNode *forward[MAX_SKIP_LIST_LEVEL] = {0};
+  bool           hasDup = false;
+  char *         pKey = NULL;
+  char *         pDataKey = NULL;
+  int            compare = 0;
+
+  tSkipListWLock(pSkipList);
 
-      tSkipListDoInsert(pSkipList, backward, pNode);
+  // backward to put the first data
+  hasDup = tSkipListGetPosToPut(pSkipList, backward, ppData[0]);
+  tSkipListPutImpl(pSkipList, ppData[0], backward, false, hasDup);
+
+  for (int level = 0; level < pSkipList->maxLevel; level++) {
+    forward[level] = SL_NODE_GET_BACKWARD_POINTER(backward[level], level);
+  }
+
+  // forward to put the rest of data
+  for (int idata = 1; idata < ndata; idata++) {
+    pDataKey = pSkipList->keyFn(ppData[idata]);
+
+    // Compare max key
+    pKey = SL_GET_MAX_KEY(pSkipList);
+    compare = pSkipList->comparFn(pDataKey, pKey);
+    if (compare > 0) {
+      for (int i = 0; i < pSkipList->maxLevel; i++) {
+        forward[i] = SL_NODE_GET_BACKWARD_POINTER(pSkipList->pTail, i);
+      }
+
+      hasDup = false;
+    } else {
+      SSkipListNode *px = pSkipList->pHead;
+      for (int i = pSkipList->maxLevel - 1; i >= 0; --i) {
+        if (i < pSkipList->level) {
+          // set new px
+          if (forward[i] != pSkipList->pHead) {
+            if (px == pSkipList->pHead ||
+                pSkipList->comparFn(SL_GET_NODE_KEY(pSkipList, forward[i]), SL_GET_NODE_KEY(pSkipList, px)) > 0) {
+              px = forward[i];
+            }
+          }
+
+          SSkipListNode *p = SL_NODE_GET_FORWARD_POINTER(px, i);
+          while (p != pSkipList->pTail) {
+            pKey = SL_GET_NODE_KEY(pSkipList, p);
+
+            compare = pSkipList->comparFn(pKey, pDataKey);
+            if (compare >= 0) {
+              if (compare == 0) hasDup = true;
+              break;
+            } else {
+              px = p;
+              p = SL_NODE_GET_FORWARD_POINTER(px, i);
+            }
+          }
+        }
+
+        forward[i] = px;
+      }
     }
+
+    tSkipListPutImpl(pSkipList, ppData[idata], forward, true, hasDup);
   }
 
   tSkipListUnlock(pSkipList);
-
-  return pNode;
 }
 
 uint32_t tSkipListRemove(SSkipList *pSkipList, SSkipListKey key) {
@@ -310,22 +367,25 @@ void tSkipListPrint(SSkipList *pSkipList, int16_t nlevel) {
   }
 }
 
-static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **backward, SSkipListNode *pNode) {
+static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **direction, SSkipListNode *pNode, bool isForward) {
   for (int32_t i = 0; i < pNode->level; ++i) {
-    if (i >= pSkipList->level) {
-      SL_NODE_GET_FORWARD_POINTER(pNode, i) = pSkipList->pTail;
-      SL_NODE_GET_BACKWARD_POINTER(pNode, i) = pSkipList->pHead;
-      SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, i) = pNode;
-      SL_NODE_GET_BACKWARD_POINTER(pSkipList->pTail, i) = pNode;
+    SSkipListNode *x = direction[i];
+    if (isForward) {
+      SL_NODE_GET_BACKWARD_POINTER(pNode, i) = x;
+
+      SSkipListNode *next = SL_NODE_GET_FORWARD_POINTER(x, i);
+      SL_NODE_GET_BACKWARD_POINTER(next, i) = pNode;
+
+      SL_NODE_GET_FORWARD_POINTER(pNode, i) = next;
+      SL_NODE_GET_FORWARD_POINTER(x, i) = pNode;
     } else {
-      SSkipListNode *x = backward[i];
       SL_NODE_GET_FORWARD_POINTER(pNode, i) = x;
 
       SSkipListNode *prev = SL_NODE_GET_BACKWARD_POINTER(x, i);
       SL_NODE_GET_FORWARD_POINTER(prev, i) = pNode;
 
-      SL_NODE_GET_BACKWARD_POINTER(x, i) = pNode;
       SL_NODE_GET_BACKWARD_POINTER(pNode, i) = prev;
+      SL_NODE_GET_BACKWARD_POINTER(x, i) = pNode;
     }
   }
 
@@ -377,7 +437,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
   char *  pDataKey = pSkipList->keyFn(pData);
 
   if (pSkipList->size == 0) {
-    for (int i = 0; i < pSkipList->level; i++) {
+    for (int i = 0; i < pSkipList->maxLevel; i++) {
       backward[i] = pSkipList->pTail;
     }
   } else {
@@ -387,7 +447,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
     pKey = SL_GET_MAX_KEY(pSkipList);
     compare = pSkipList->comparFn(pDataKey, pKey);
     if (compare >= 0) {
-      for (int i = 0; i < pSkipList->level; i++) {
+      for (int i = 0; i < pSkipList->maxLevel; i++) {
         backward[i] = pSkipList->pTail;
       }
 
@@ -398,7 +458,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
     pKey = SL_GET_MIN_KEY(pSkipList);
     compare = pSkipList->comparFn(pDataKey, pKey);
     if (compare < 0) {
-      for (int i = 0; i < pSkipList->level; i++) {
+      for (int i = 0; i < pSkipList->maxLevel; i++) {
         backward[i] = SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, i);
       }
 
@@ -406,18 +466,20 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
     }
 
     SSkipListNode *px = pSkipList->pTail;
-    for (int i = pSkipList->level - 1; i >= 0; --i) {
-      SSkipListNode *p = SL_NODE_GET_BACKWARD_POINTER(px, i);
-      while (p != pSkipList->pHead) {
-        pKey = SL_GET_NODE_KEY(pSkipList, p);
-
-        compare = pSkipList->comparFn(pKey, pDataKey);
-        if (compare <= 0) {
-          if (compare == 0 && !hasDupKey) hasDupKey = true;
-          break;
-        } else {
-          px = p;
-          p = SL_NODE_GET_BACKWARD_POINTER(px, i);
+    for (int i = pSkipList->maxLevel - 1; i >= 0; --i) {
+      if (i < pSkipList->level) {
+        SSkipListNode *p = SL_NODE_GET_BACKWARD_POINTER(px, i);
+        while (p != pSkipList->pHead) {
+          pKey = SL_GET_NODE_KEY(pSkipList, p);
+
+          compare = pSkipList->comparFn(pKey, pDataKey);
+          if (compare <= 0) {
+            if (compare == 0 && !hasDupKey) hasDupKey = true;
+            break;
+          } else {
+            px = p;
+            p = SL_NODE_GET_BACKWARD_POINTER(px, i);
+          }
         }
       }
 
@@ -579,6 +641,32 @@ static SSkipListNode *tSkipListNewNode(uint8_t level) {
   return pNode;
 }
 
+static SSkipListNode *tSkipListPutImpl(SSkipList *pSkipList, void *pData, SSkipListNode **direction, bool isForward,
+                                       bool hasDup) {
+  uint8_t        dupMode = SL_DUP_MODE(pSkipList);
+  SSkipListNode *pNode = NULL;
+
+  if (hasDup && (dupMode == SL_DISCARD_DUP_KEY || dupMode == SL_UPDATE_DUP_KEY)) {
+    if (dupMode == SL_UPDATE_DUP_KEY) {
+      if (isForward) {
+        pNode = SL_NODE_GET_FORWARD_POINTER(direction[0], 0);
+      } else {
+        pNode = SL_NODE_GET_BACKWARD_POINTER(direction[0], 0);
+      }
+      atomic_store_ptr(&(pNode->pData), pData);
+    }
+  } else {
+    pNode = tSkipListNewNode(getSkipListRandLevel(pSkipList));
+    if (pNode != NULL) {
+      pNode->pData = pData;
+
+      tSkipListDoInsert(pSkipList, direction, pNode, isForward);
+    }
+  }
+
+  return pNode;
+}
+
 // static int32_t tSkipListEndParQuery(SSkipList *pSkipList, SSkipListNode *pStartNode, SSkipListKey *pEndKey,
 //                                    int32_t cond, SSkipListNode ***pRes) {
 //  pthread_rwlock_rdlock(&pSkipList->lock);