Merge pull request #2433 from taosdata/feature/2.0tsdb

Feature/2.0tsdb

src/tsdb/src/tsdbMain.c

@@ -179,7 +179,7 @@ int32_t tsdbInsertData(TSDB_REPO_T *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *
       return -1;
     }
   }
-  pRsp->affectedRows = htonl(affectedrows);
+  if (pRsp != NULL) pRsp->affectedRows = htonl(affectedrows);
   return 0;
 }
 
@@ -648,7 +648,7 @@ static STsdbRepo *tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg) {
   }
 
   pRepo->config = *pCfg;
-  pRepo->appH = *pAppH;
+  if (pAppH) pRepo->appH = *pAppH;
 
   pRepo->tsdbMeta = tsdbNewMeta(pCfg);
   if (pRepo->tsdbMeta == NULL) {

src/tsdb/src/tsdbRWHelper.c

@@ -173,9 +173,14 @@ int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
       close(pHelper->files.nHeadF.fd);
       pHelper->files.nHeadF.fd = -1;
       if (hasError) {
-        remove(pHelper->files.nHeadF.fname);
+        (void)remove(pHelper->files.nHeadF.fname);
       } else {
-        rename(pHelper->files.nHeadF.fname, pHelper->files.headF.fname);
+        if (rename(pHelper->files.nHeadF.fname, pHelper->files.headF.fname) < 0) {
+          tsdbError("vgId:%d failed to rename file from %s to %s since %s", REPO_ID(pHelper->pRepo),
+                    pHelper->files.nHeadF.fname, pHelper->files.headF.fname, strerror(errno));
+          terrno = TAOS_SYSTEM_ERROR(errno);
+          return -1;
+        }
         pHelper->files.headF.info = pHelper->files.nHeadF.info;
       }
     }
@@ -186,9 +191,14 @@ int tsdbCloseHelperFile(SRWHelper *pHelper, bool hasError) {
       close(pHelper->files.nLastF.fd);
       pHelper->files.nLastF.fd = -1;
       if (hasError) {
-        remove(pHelper->files.nLastF.fname);
+        (void)remove(pHelper->files.nLastF.fname);
       } else {
-        rename(pHelper->files.nLastF.fname, pHelper->files.lastF.fname);
+        if (rename(pHelper->files.nLastF.fname, pHelper->files.lastF.fname) < 0) {
+          tsdbError("vgId:%d failed to rename file from %s to %s since %s", REPO_ID(pHelper->pRepo),
+                    pHelper->files.nLastF.fname, pHelper->files.lastF.fname, strerror(errno));
+          terrno = TAOS_SYSTEM_ERROR(errno);
+          return -1;
+        }
         pHelper->files.lastF.info = pHelper->files.nLastF.info;
       }
     }
@@ -306,8 +316,7 @@ int tsdbMoveLastBlockIfNeccessary(SRWHelper *pHelper) {
 
     if (pCompBlock->numOfSubBlocks > 1) {
       if (tsdbLoadBlockData(pHelper, blockAtIdx(pHelper, pIdx->numOfBlocks - 1), NULL) < 0) return -1;
-      ASSERT(pHelper->pDataCols[0]->numOfRows > 0 &&
-             pHelper->pDataCols[0]->numOfRows < pCfg->minRowsPerFileBlock);
+      ASSERT(pHelper->pDataCols[0]->numOfRows > 0 && pHelper->pDataCols[0]->numOfRows < pCfg->minRowsPerFileBlock);
       if (tsdbWriteBlockToFile(pHelper, &(pHelper->files.nLastF), pHelper->pDataCols[0],
                                pHelper->pDataCols[0]->numOfRows, &compBlock, true, true) < 0)
         return -1;
@@ -330,14 +339,27 @@ int tsdbMoveLastBlockIfNeccessary(SRWHelper *pHelper) {
 }
 
 int tsdbWriteCompInfo(SRWHelper *pHelper) {
+  off_t offset = 0;
   SCompIdx *pIdx = pHelper->pCompIdx + pHelper->tableInfo.tid;
   if (!helperHasState(pHelper, TSDB_HELPER_INFO_LOAD)) {
     if (pIdx->offset > 0) {
-      pIdx->offset = lseek(pHelper->files.nHeadF.fd, 0, SEEK_END);
-      if (pIdx->offset < 0) return -1;
+      offset = lseek(pHelper->files.nHeadF.fd, 0, SEEK_END);
+      if (offset < 0) {
+        tsdbError("vgId:%d failed to lseed file %s since %s", REPO_ID(pHelper->pRepo), pHelper->files.nHeadF.fname,
+                  strerror(errno));
+        terrno = TAOS_SYSTEM_ERROR(errno);
+        return -1;
+      }
+
+      pIdx->offset = offset;
       ASSERT(pIdx->offset >= TSDB_FILE_HEAD_SIZE);
 
-      if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, pIdx->len) < pIdx->len) return -1;
+      if (tsendfile(pHelper->files.nHeadF.fd, pHelper->files.headF.fd, NULL, pIdx->len) < pIdx->len) {
+        tsdbError("vgId:%d failed to send %d bytes from file %s to %s since %s", REPO_ID(pHelper->pRepo), pIdx->len,
+                  pHelper->files.headF.fname, pHelper->files.nHeadF.fname, strerror(errno));
+        terrno = TAOS_SYSTEM_ERROR(errno);
+        return -1;
+      }
     }
   } else {
     pHelper->pCompInfo->delimiter = TSDB_FILE_DELIMITER;
@@ -345,12 +367,23 @@ int tsdbWriteCompInfo(SRWHelper *pHelper) {
     pHelper->pCompInfo->checksum = 0;
     ASSERT((pIdx->len - sizeof(SCompInfo) - sizeof(TSCKSUM)) % sizeof(SCompBlock) == 0);
     taosCalcChecksumAppend(0, (uint8_t *)pHelper->pCompInfo, pIdx->len);
-    pIdx->offset = lseek(pHelper->files.nHeadF.fd, 0, SEEK_END);
+    offset = lseek(pHelper->files.nHeadF.fd, 0, SEEK_END);
+    if (offset < 0) {
+      tsdbError("vgId:%d failed to lseek file %s since %s", REPO_ID(pHelper->pRepo), pHelper->files.nHeadF.fname,
+                strerror(errno));
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      return -1;
+    }
+    pIdx->offset = offset;
     pIdx->uid = pHelper->tableInfo.uid;
-    if (pIdx->offset < 0) return -1;
     ASSERT(pIdx->offset >= TSDB_FILE_HEAD_SIZE);
 
-    if (twrite(pHelper->files.nHeadF.fd, (void *)(pHelper->pCompInfo), pIdx->len) < pIdx->len) return -1;
+    if (twrite(pHelper->files.nHeadF.fd, (void *)(pHelper->pCompInfo), pIdx->len) < pIdx->len) {
+      tsdbError("vgId:%d failed to write %d bytes to file %s since %s", REPO_ID(pHelper->pRepo), pIdx->len,
+                pHelper->files.nHeadF.fname, strerror(errno));
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      return -1;
+    }
   }
 
   return 0;
@@ -567,24 +600,24 @@ _err:
 static bool tsdbShouldCreateNewLast(SRWHelper *pHelper) {
   ASSERT(pHelper->files.lastF.fd > 0);
   struct stat st;
-  fstat(pHelper->files.lastF.fd, &st);
+  if (fstat(pHelper->files.lastF.fd, &st) < 0) return true;
   if (st.st_size > 32 * 1024 + TSDB_FILE_HEAD_SIZE) return true;
   return false;
 }
 
 static int tsdbWriteBlockToFile(SRWHelper *pHelper, SFile *pFile, SDataCols *pDataCols, int rowsToWrite,
                                 SCompBlock *pCompBlock, bool isLast, bool isSuperBlock) {
-  STsdbCfg *pCfg = &(pHelper->pRepo->config);
+  STsdbCfg * pCfg = &(pHelper->pRepo->config);
   SCompData *pCompData = (SCompData *)(pHelper->pBuffer);
   int64_t    offset = 0;
 
   ASSERT(rowsToWrite > 0 && rowsToWrite <= pDataCols->numOfRows && rowsToWrite <= pCfg->maxRowsPerFileBlock);
   ASSERT(isLast ? rowsToWrite < pCfg->minRowsPerFileBlock : true);
 
-
   offset = lseek(pFile->fd, 0, SEEK_END);
   if (offset < 0) {
-    tsdbError("vgId:%d failed to write block to file %s since %s", REPO_ID(pHelper->pRepo), pFile->fname, strerror(errno));
+    tsdbError("vgId:%d failed to write block to file %s since %s", REPO_ID(pHelper->pRepo), pFile->fname,
+              strerror(errno));
     terrno = TAOS_SYSTEM_ERROR(errno);
     goto _err;
   }
@@ -639,9 +672,9 @@ static int tsdbWriteBlockToFile(SRWHelper *pHelper, SFile *pFile, SDataCols *pDa
         }
       }
 
-      pCompCol->len = (*(tDataTypeDesc[pDataCol->type].compFunc))(
-          (char *)pDataCol->pData, tlen, rowsToWrite, tptr, tsizeof(pHelper->pBuffer) - lsize, pCfg->compression,
-          pHelper->compBuffer, tsizeof(pHelper->compBuffer));
+      pCompCol->len = (*(tDataTypeDesc[pDataCol->type].compFunc))((char *)pDataCol->pData, tlen, rowsToWrite, tptr,
+                                                                  tsizeof(pHelper->pBuffer) - lsize, pCfg->compression,
+                                                                  pHelper->compBuffer, tsizeof(pHelper->compBuffer));
     } else {
       pCompCol->len = tlen;
       memcpy(tptr, pDataCol->pData, pCompCol->len);
@@ -725,8 +758,7 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
   // ASSERT(compareKeyBlock((void *)&keyFirst, (void *)pCompBlock) == 0);
 
   if (keyFirst > blockAtIdx(pHelper, blkIdx)->keyLast) {  // Merge with the last block by append
-    ASSERT(blockAtIdx(pHelper, blkIdx)->numOfRows < pCfg->minRowsPerFileBlock &&
-           blkIdx == pIdx->numOfBlocks - 1);
+    ASSERT(blockAtIdx(pHelper, blkIdx)->numOfRows < pCfg->minRowsPerFileBlock && blkIdx == pIdx->numOfBlocks - 1);
     int defaultRowsToWrite = pCfg->maxRowsPerFileBlock * 4 / 5;  // TODO: make a interface
 
     rowsWritten = MIN((defaultRowsToWrite - blockAtIdx(pHelper, blkIdx)->numOfRows), pDataCols->numOfRows);
@@ -1051,7 +1083,7 @@ static void tsdbResetHelperFileImpl(SRWHelper *pHelper) {
 
 static int tsdbInitHelperFile(SRWHelper *pHelper) {
   STsdbCfg *pCfg = &pHelper->pRepo->config;
-  size_t tsize = sizeof(SCompIdx) * pCfg->maxTables + sizeof(TSCKSUM);
+  size_t    tsize = sizeof(SCompIdx) * pCfg->maxTables + sizeof(TSCKSUM);
   pHelper->pCompIdx = (SCompIdx *)tmalloc(tsize);
   if (pHelper->pCompIdx == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
@@ -1099,10 +1131,8 @@ static int tsdbInitHelperBlock(SRWHelper *pHelper) {
   STsdbRepo *pRepo = helperRepo(pHelper);
   STsdbMeta *pMeta = pHelper->pRepo->tsdbMeta;
 
-  pHelper->pDataCols[0] =
-      tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
-  pHelper->pDataCols[1] =
-      tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
+  pHelper->pDataCols[0] = tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
+  pHelper->pDataCols[1] = tdNewDataCols(pMeta->maxRowBytes, pMeta->maxCols, pRepo->config.maxRowsPerFileBlock);
   if (pHelper->pDataCols[0] == NULL || pHelper->pDataCols[1] == NULL) {
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return -1;
@@ -1222,12 +1252,16 @@ static int tsdbCheckAndDecodeColumnData(SDataCol *pDataCol, char *content, int32
 static int tsdbLoadBlockDataImpl(SRWHelper *pHelper, SCompBlock *pCompBlock, SDataCols *pDataCols) {
   ASSERT(pCompBlock->numOfSubBlocks <= 1);
 
-  ASSERT(tsizeof(pHelper->pBuffer) >= pCompBlock->len);
-
   SCompData *pCompData = (SCompData *)pHelper->pBuffer;
 
   SFile *pFile = (pCompBlock->last) ? &(pHelper->files.lastF) : &(pHelper->files.dataF);
 
+  pHelper->pBuffer = trealloc(pHelper->pBuffer, pCompBlock->len);
+  if (pHelper->pBuffer == NULL) {
+    terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
+    goto _err;
+  }
+
   int fd = pFile->fd;
   if (lseek(fd, pCompBlock->offset, SEEK_SET) < 0) {
     tsdbError("vgId:%d tid:%d failed to lseek file %s since %s", REPO_ID(pHelper->pRepo), pHelper->tableInfo.tid,

src/tsdb/tests/CMakeLists.txt

 aux_source_directory(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 
 add_executable(tsdbTests ${SOURCE_LIST})
-target_link_libraries(tsdbTests gtest gtest_main pthread common tsdb)
+target_link_libraries(tsdbTests gtest gtest_main pthread common tsdb tutil trpc)
 
 add_test(NAME unit COMMAND ${CMAKE_CURRENT_BINARY_DIR}/tsdbTests)
\ No newline at end of file

src/tsdb/tests/tsdbTests.cpp

@@ -2,9 +2,8 @@
 #include <stdlib.h>
 #include <sys/time.h>
 
-#include "tdataformat.h"
+#include "tsdb.h"
 #include "tsdbMain.h"
-#include "tskiplist.h"
 
 static double getCurTime() {
   struct timeval tv;
@@ -77,7 +76,7 @@ static int insertData(SInsertInfo *pInfo) {
     pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
     pMsg->compressed = htonl(pMsg->numOfBlocks);
 
-    if (tsdbInsertData(pInfo->pRepo, pMsg) < 0) {
+    if (tsdbInsertData(pInfo->pRepo, pMsg, NULL) < 0) {
       tfree(pMsg);
       return -1;
     }
@@ -90,222 +89,72 @@ static int insertData(SInsertInfo *pInfo) {
   return 0;
 }
 
-TEST(TsdbTest, DISABLED_tableEncodeDecode) {
-// TEST(TsdbTest, tableEncodeDecode) {
-  STable *pTable = (STable *)malloc(sizeof(STable));
-
-  pTable->type = TSDB_NORMAL_TABLE;
-  pTable->tableId.uid = 987607499877672L;
-  pTable->tableId.tid = 0;
-  pTable->superUid = -1;
-  pTable->sversion = 0;
-  pTable->tagSchema = NULL;
-  pTable->tagVal = NULL;
-  int nCols = 5;
-  STSchema *schema = tdNewSchema(nCols);
-
-  for (int i = 0; i < nCols; i++) {
-    if (i == 0) {
-      tdSchemaAddCol(schema, TSDB_DATA_TYPE_TIMESTAMP, i, -1);
-    } else {
-      tdSchemaAddCol(schema, TSDB_DATA_TYPE_INT, i, -1);
-    }
-  }
-
-  pTable->schema = schema;
-
-  int bufLen = 0;
-  void *buf = tsdbEncodeTable(pTable, &bufLen);
-
-  STable *tTable = tsdbDecodeTable(buf, bufLen);
-
-  ASSERT_EQ(pTable->type, tTable->type);
-  ASSERT_EQ(pTable->tableId.uid, tTable->tableId.uid);
-  ASSERT_EQ(pTable->tableId.tid, tTable->tableId.tid);
-  ASSERT_EQ(pTable->superUid, tTable->superUid);
-  ASSERT_EQ(pTable->sversion, tTable->sversion);
-  ASSERT_EQ(memcmp(pTable->schema, tTable->schema, sizeof(STSchema) + sizeof(STColumn) * nCols), 0);
+static void tsdbSetCfg(STsdbCfg *pCfg, int32_t tsdbId, int32_t cacheBlockSize, int32_t totalBlocks, int32_t maxTables,
+                       int32_t daysPerFile, int32_t keep, int32_t minRows, int32_t maxRows, int8_t precision,
+                       int8_t compression) {
+  pCfg->tsdbId = tsdbId;
+  pCfg->cacheBlockSize = cacheBlockSize;
+  pCfg->totalBlocks = totalBlocks;
+  pCfg->maxTables = maxTables;
+  pCfg->daysPerFile = daysPerFile;
+  pCfg->keep = keep;
+  pCfg->minRowsPerFileBlock = minRows;
+  pCfg->maxRowsPerFileBlock = maxRows;
+  pCfg->precision = precision;
+  pCfg->compression = compression;
 }
 
-// TEST(TsdbTest, DISABLED_createRepo) {
-TEST(TsdbTest, createRepo) {
-  STsdbCfg config;
-  STsdbRepo *repo;
-
-  // 1. Create a tsdb repository
-  tsdbSetDefaultCfg(&config);
-  ASSERT_EQ(tsdbCreateRepo("/home/ubuntu/work/ttest/vnode0", &config, NULL), 0);
-
-  TSDB_REPO_T *pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
-  ASSERT_NE(pRepo, nullptr);
+static void tsdbSetTableCfg(STableCfg *pCfg) {
+  STSchemaBuilder schemaBuilder = {0};
 
-  // 2. Create a normal table
-  STableCfg tCfg;
-  ASSERT_EQ(tsdbInitTableCfg(&tCfg, TSDB_SUPER_TABLE, 987607499877672L, 0), -1);
-  ASSERT_EQ(tsdbInitTableCfg(&tCfg, TSDB_NORMAL_TABLE, 987607499877672L, 0), 0);
-  tsdbTableSetName(&tCfg, "test", false);
+  pCfg->type = TSDB_NORMAL_TABLE;
+  pCfg->superUid = TSDB_INVALID_SUPER_TABLE_ID;
+  pCfg->tableId.tid = 1;
+  pCfg->tableId.uid = 5849583783847394;
+  tdInitTSchemaBuilder(&schemaBuilder, 0);
 
-  int       nCols = 5;
-  STSchema *schema = tdNewSchema(nCols);
-
-  for (int i = 0; i < nCols; i++) {
-    if (i == 0) {
-      tdSchemaAddCol(schema, TSDB_DATA_TYPE_TIMESTAMP, i, -1);
-    } else {
-      tdSchemaAddCol(schema, TSDB_DATA_TYPE_INT, i, -1);
-    }
+  int colId = 0;
+  for (int i = 0; i < 5; i++) {
+    tdAddColToSchema(&schemaBuilder, (colId == 0) ? TSDB_DATA_TYPE_TIMESTAMP : TSDB_DATA_TYPE_INT, colId, 0);
+    colId++;
   }
 
-  tsdbTableSetSchema(&tCfg, schema, true);
-
-  tsdbCreateTable(pRepo, &tCfg);
+  pCfg->schema = tdGetSchemaFromBuilder(&schemaBuilder);
+  pCfg->name = strdup("t1");
 
-  // Insert Some Data
-  SInsertInfo iInfo = {
-    .pRepo = pRepo,
-    // .isAscend = true,
-    .isAscend = false,
-    .tid = tCfg.tableId.tid,
-    .uid = tCfg.tableId.uid,
-    .sversion = tCfg.sversion,
-    .startTime = 1584081000000,
-    .interval = 1000,
-    .totalRows = 10000000,
-    .rowsPerSubmit = 1,
-    .pSchema = schema
-  };
-
-  ASSERT_EQ(insertData(&iInfo), 0);
-
-  // Close the repository
-  tsdbCloseRepo(pRepo);
-
-  // Open the repository again
-  pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
-  repo = (STsdbRepo *)pRepo;
-  ASSERT_NE(pRepo, nullptr);
-
-  // // Insert more data
-  // iInfo.startTime = iInfo.startTime + iInfo.interval * iInfo.totalRows;
-  // iInfo.totalRows = 10;
-  // iInfo.pRepo = pRepo;
-  // ASSERT_EQ(insertData(&iInfo), 0);
-
-  // // Close the repository
-  // tsdbCloseRepo(pRepo);
-
-  // // Open the repository again
-  // pRepo = tsdbOpenRepo("/home/ubuntu/work/ttest/vnode0", NULL);
-  // repo = (STsdbRepo *)pRepo;
-  // ASSERT_NE(pRepo, nullptr);
-
-  // // Read from file
-  // SRWHelper rhelper;
-  // tsdbInitReadHelper(&rhelper, repo);
-
-  // SFileGroup *pFGroup = tsdbSearchFGroup(repo->tsdbFileH, 1833);
-  // ASSERT_NE(pFGroup, nullptr);
-  // ASSERT_GE(tsdbSetAndOpenHelperFile(&rhelper, pFGroup), 0);
-
-  // STable *pTable = tsdbGetTableByUid(repo->tsdbMeta, tCfg.tableId.uid);
-  // ASSERT_NE(pTable, nullptr);
-  // tsdbSetHelperTable(&rhelper, pTable, repo);
-
-  // ASSERT_EQ(tsdbLoadCompInfo(&rhelper, NULL), 0);
-  // ASSERT_EQ(tsdbLoadBlockData(&rhelper, blockAtIdx(&rhelper, 0), NULL), 0);
-
-  int k = 0;
+  tdDestroyTSchemaBuilder(&schemaBuilder);
 }
 
-TEST(TsdbTest, DISABLED_openRepo) {
-// TEST(TsdbTest, openRepo) {
-  // tsdb_repo_t *repo = tsdbOpenRepo("/home/ubuntu/work/build/test/data/vnode/vnode2/tsdb", NULL);
-  // ASSERT_NE(repo, nullptr);
-
-  // STsdbRepo *pRepo = (STsdbRepo *)repo;
-
-  // SFileGroup *pGroup = tsdbSearchFGroup(pRepo->tsdbFileH, 1655);
+TEST(TsdbTest, testInsertSpeed) {
+  int         vnode = 1;
+  int         ret = 0;
+  STsdbCfg    tsdbCfg;
+  STableCfg   tableCfg;
+  std::string testDir = "./test";
+  char *      rootDir = strdup((testDir + "/vnode" + std::to_string(vnode)).c_str());
 
-//   for (int type = TSDB_FILE_TYPE_HEAD; type < TSDB_FILE_TYPE_MAX; type++) {
-//     tsdbOpenFile(&pGroup->files[type], O_RDONLY);
-//   }
+  tsdbDebugFlag = 131; //NOTE: you must set the flag
 
-//   SCompIdx *pIdx = (SCompIdx *)calloc(pRepo->config.maxTables, sizeof(SCompIdx));
-//   tsdbLoadCompIdx(pGroup, (void *)pIdx, pRepo->config.maxTables);
+  taosRemoveDir(rootDir);
 
-//   SCompInfo *pCompInfo = (SCompInfo *)malloc(sizeof(SCompInfo) + pIdx[1].len);
+  // Create and open repository
+  tsdbSetCfg(&tsdbCfg, 1, 16, 4, -1, -1, -1, -1, -1, -1, -1);
+  tsdbCreateRepo(rootDir, &tsdbCfg);
+  TSDB_REPO_T *repo = tsdbOpenRepo(rootDir, NULL);
+  ASSERT_NE(repo, nullptr);
 
-  // tsdbLoadCompBlocks(pGroup, &pIdx[1], (void *)pCompInfo);
+  // Create table
+  tsdbSetTableCfg(&tableCfg);
+  tsdbCreateTable(repo, &tableCfg);
 
-//   int blockIdx = 0;
-//   SCompBlock *pBlock = &(pCompInfo->blocks[blockIdx]);
+  // Insert data
+  SInsertInfo iInfo = {repo, true, 1, 5849583783847394, 0, 1590000000000, 10, 10000000, 100, tableCfg.schema};
 
-//   SCompData *pCompData = (SCompData *)malloc(sizeof(SCompData) + sizeof(SCompCol) * pBlock->numOfCols);
-
-//   tsdbLoadCompCols(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock, (void *)pCompData);
-
-  // STable *pTable = tsdbGetTableByUid(pRepo->tsdbMeta, pCompData->uid);
-  // SDataCols *pDataCols = tdNewDataCols(tdMaxRowBytesFromSchema(tsdbGetTableSchema(pRepo->tsdbMeta, pTable)), 5);
-  // tdInitDataCols(pDataCols, tsdbGetTableSchema(pRepo->tsdbMeta, pTable));
-
-//   tsdbLoadDataBlock(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock, 1, pDataCols, pCompData);
-
-  // tdResetDataCols(pDataCols);
-
-  // tsdbLoadDataBlock(&pGroup->files[TSDB_FILE_TYPE_DATA], pBlock + 1, 1, pDataCols, pCompData);
-
-
-//   int k = 0;
+  insertData(&iInfo);
 
-}
-
-TEST(TsdbTest, DISABLED_createFileGroup) {
-  SFileGroup fGroup;
-
-  // ASSERT_EQ(tsdbCreateFileGroup("/home/ubuntu/work/ttest/vnode0/data", 1820, &fGroup, 1000), 0);
-
-  int k = 0;
+  tsdbCloseRepo(repo, 1);
 }
 
 static char *getTKey(const void *data) {
   return (char *)data;
-}
-
-static void insertSkipList(bool isAscend) {
-  TSKEY start_time = 1587393453000;
-  TSKEY interval = 1000;
-
-  SSkipList *pList = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, sizeof(TSKEY), 0, 0, 1, getTKey);
-  ASSERT_NE(pList, nullptr);
-
-  for (size_t i = 0; i < 20000000; i++)
-  {
-    TSKEY time = isAscend ? (start_time + i * interval) : (start_time - i * interval);
-    int32_t level = 0;
-    int32_t headSize = 0;
-
-    tSkipListNewNodeInfo(pList, &level, &headSize);
-    SSkipListNode *pNode = (SSkipListNode *)malloc(headSize + sizeof(TSKEY));
-    ASSERT_NE(pNode, nullptr);
-    pNode->level = level;
-    *(TSKEY *)((char *)pNode + headSize) = time;
-    tSkipListPut(pList, pNode);
-  }
-
-  tSkipListDestroy(pList);
-}
-
-TEST(TsdbTest, DISABLED_testSkipList) {
-// TEST(TsdbTest, testSkipList) {
-  double stime = getCurTime();
-  insertSkipList(true);
-  double etime = getCurTime();
-
-  printf("Time used to insert 100000000 records takes %f seconds\n", etime-stime);
-
-  stime = getCurTime();
-  insertSkipList(false);
-  etime = getCurTime();
-
-  printf("Time used to insert 100000000 records takes %f seconds\n", etime-stime);
 }
\ No newline at end of file

src/util/src/tkvstore.c

@@ -78,8 +78,8 @@ int tdCreateKVStore(char *fname) {
   return 0;
 
 _err:
-  if (fd > 0) close(fd);
-  remove(fname);
+  if (fd >= 0) close(fd);
+  (void)remove(fname);
   return -1;
 }
 
@@ -106,15 +106,15 @@ SKVStore *tdOpenKVStore(char *fname, iterFunc iFunc, afterFunc aFunc, void *appH
     goto _err;
   }
 
-  if (access(pStore->fsnap, F_OK) == 0) { // .snap file exists
-    uTrace("file %s exists, try to recover the KV store", pStore->fsnap);
-    pStore->sfd = open(pStore->fsnap, O_RDONLY);
-    if (pStore->sfd < 0) {
+  pStore->sfd = open(pStore->fsnap, O_RDONLY);
+  if (pStore->sfd < 0) {
+    if (errno != ENOENT) {
       uError("failed to open file %s since %s", pStore->fsnap, strerror(errno));
       terrno = TAOS_SYSTEM_ERROR(errno);
       goto _err;
     }
-
+  } else {
+    uTrace("file %s exists, try to recover the KV store", pStore->fsnap);
     if (tdLoadKVStoreHeader(pStore->sfd, pStore->fsnap, &info) < 0) {
       if (terrno != TSDB_CODE_COM_FILE_CORRUPTED) goto _err;
     } else {
@@ -133,7 +133,7 @@ SKVStore *tdOpenKVStore(char *fname, iterFunc iFunc, afterFunc aFunc, void *appH
 
     close(pStore->sfd);
     pStore->sfd = -1;
-    remove(pStore->fsnap);
+    (void)remove(pStore->fsnap);
   }
 
   if (tdLoadKVStoreHeader(pStore->fd, pStore->fname, &info) < 0) goto _err;
@@ -212,7 +212,7 @@ _err:
   if (pStore->sfd > 0) {
     close(pStore->sfd);
     pStore->sfd = -1;
-    remove(pStore->fsnap);
+    (void)remove(pStore->fsnap);
   }
   if (pStore->fd > 0) {
     close(pStore->fd);
@@ -314,7 +314,7 @@ int tdKVStoreEndCommit(SKVStore *pStore) {
   }
   pStore->fd = -1;
 
-  remove(pStore->fsnap);
+  (void)remove(pStore->fsnap);
   return 0;
 }