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

Feature/2.0tsdb

src/vnode/tsdb/src/tsdbCache.c

@@ -17,8 +17,9 @@
 #include "tsdb.h"
 #include "tsdbCache.h"
 
-static int tsdbAllocBlockFromPool(STsdbCache *pCache);
-static void tsdbFreeBlockList(SCacheMem *mem);
+static int  tsdbAllocBlockFromPool(STsdbCache *pCache);
+static void tsdbFreeBlockList(SList *list);
+static void tsdbFreeCacheMem(SCacheMem *mem);
 
 STsdbCache *tsdbInitCache(int maxBytes, int cacheBlockSize, tsdb_repo_t *pRepo) {
   STsdbCache *pCache = (STsdbCache *)calloc(1, sizeof(STsdbCache));
@@ -60,8 +61,8 @@ _err:
 }
 
 void tsdbFreeCache(STsdbCache *pCache) {
-  tsdbFreeBlockList(pCache->imem);
-  tsdbFreeBlockList(pCache->mem);
+  tsdbFreeCacheMem(pCache->imem);
+  tsdbFreeCacheMem(pCache->mem);
   tsdbFreeBlockList(pCache->pool.memPool);
   free(pCache);
 }
@@ -90,9 +91,7 @@ void *tsdbAllocFromCache(STsdbCache *pCache, int bytes, TSKEY key) {
   return ptr;
 }
 
-static void tsdbFreeBlockList(SCacheMem *mem) {
-  if (mem == NULL) return;
-  SList *          list = mem->list;
+static void tsdbFreeBlockList(SList *list) {
   SListNode *      node = NULL;
   STsdbCacheBlock *pBlock = NULL;
   while ((node = tdListPopHead(list)) != NULL) {
@@ -101,6 +100,12 @@ static void tsdbFreeBlockList(SCacheMem *mem) {
     listNodeFree(node);
   }
   tdListFree(list);
+}
+
+static void tsdbFreeCacheMem(SCacheMem *mem) {
+  if (mem == NULL) return;
+  SList *list = mem->list;
+  tsdbFreeBlockList(list);
   free(mem);
 }
 

src/vnode/tsdb/src/tsdbMain.c

@@ -712,7 +712,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
   tSkipListRandNodeInfo(pTable->mem->pData, &level, &headSize);
 
   TSKEY key = dataRowKey(row);
-  printf("insert:%lld, size:%d\n", key, pTable->mem->numOfPoints);
+  // printf("insert:%lld, size:%d\n", key, pTable->mem->numOfPoints);
   
   // Copy row into the memory
   SSkipListNode *pNode = tsdbAllocFromCache(pRepo->tsdbCache, headSize + dataRowLen(row), key);

src/vnode/tsdb/tests/tsdbTests.cpp

@@ -79,7 +79,7 @@ TEST(TsdbTest, createRepo) {
 
   // // 3. Loop to write some simple data
   int nRows = 10000000;
-  int rowsPerSubmit = 100;
+  int rowsPerSubmit = 10;
   int64_t start_time = 1584081000000;
 
   SSubmitMsg *pMsg = (SSubmitMsg *)malloc(sizeof(SSubmitMsg) + sizeof(SSubmitBlk) + tdMaxRowBytesFromSchema(schema) * rowsPerSubmit);
@@ -88,7 +88,8 @@ TEST(TsdbTest, createRepo) {
 
   for (int k = 0; k < nRows/rowsPerSubmit; k++) {
     SSubmitBlk *pBlock = pMsg->blocks;
-    pBlock->tableId = {.uid = 987607499877672L, .tid = 0};
+    pBlock->uid = 987607499877672L;
+    pBlock->tid = 0;
     pBlock->sversion = 0;
     pBlock->len = 0;
     for (int i = 0; i < rowsPerSubmit; i++) {
@@ -107,7 +108,18 @@ TEST(TsdbTest, createRepo) {
       }
       pBlock->len += dataRowLen(row);
     }
+    pBlock->len = htonl(pBlock->len);
+    pBlock->numOfRows = htonl(pBlock->numOfRows);
+    pBlock->uid = htobe64(pBlock->uid);
+    pBlock->tid = htonl(pBlock->tid);
+
+    pBlock->sversion = htonl(pBlock->sversion);
+    pBlock->padding = htonl(pBlock->padding);
+
     pMsg->length = pMsg->length + sizeof(SSubmitBlk) + pBlock->len;
+    pMsg->length = htonl(pMsg->length);
+    pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
+    pMsg->compressed = htonl(pMsg->numOfBlocks);
 
     tsdbInsertData(pRepo, pMsg);
   }