TD-34

src/common/inc/dataformat.h

@@ -81,11 +81,13 @@ STSchema *tdDecodeSchema(void **psrc);
  */
 typedef void *SDataRow;
 
+
 #define TD_DATA_ROW_HEAD_SIZE (2 * sizeof(int32_t))
 
 #define dataRowLen(r) (*(int32_t *)(r))
 #define dataRowFLen(r) (*(int32_t *)((char *)(r) + sizeof(int32_t)))
 #define dataRowTuple(r) ((char *)(r) + TD_DATA_ROW_HEAD_SIZE)
+#define dataRowKey(r) (*(TSKEY *)(dataRowTuple(r)))
 #define dataRowSetLen(r, l) (dataRowLen(r) = (l))
 #define dataRowSetFLen(r, l) (dataRowFLen(r) = (l))
 #define dataRowIdx(r, i) ((char *)(r) + i)

src/vnode/tsdb/inc/tsdbMeta.h

@@ -33,20 +33,25 @@ extern "C" {
 
 #define IS_CREATE_STABLE(pCfg) ((pCfg)->tagValues != NULL)
 
+typedef struct {
+  TSKEY   keyFirst;
+  TSKEY   keyLast;
+  int32_t numOfPoints;
+  void *  pData;
+} SMemTable;
+
 // ---------- TSDB TABLE DEFINITION
 typedef struct STable {
-  int8_t    type;
-  STableId  tableId;
-  int32_t   superUid;  // Super table UID
-  int32_t   sversion;
-  STSchema *schema;
-  STSchema *tagSchema;
-  SDataRow  tagVal;
-  union {
-    void *pData;   // For TSDB_NORMAL_TABLE and TSDB_CHILD_TABLE, it is the skiplist for cache data
-    void *pIndex;  // For TSDB_SUPER_TABLE, it is the skiplist index
-  } content;
-  void *         iData;          // Skiplist to commit
+  int8_t         type;
+  STableId       tableId;
+  int32_t        superUid;  // Super table UID
+  int32_t        sversion;
+  STSchema *     schema;
+  STSchema *     tagSchema;
+  SDataRow       tagVal;
+  SMemTable *    mem;
+  SMemTable *    imem;
+  void *         pIndex;         // For TSDB_SUPER_TABLE, it is the skiplist index
   void *         eventHandler;   // TODO
   void *         streamHandler;  // TODO
   struct STable *next;           // TODO: remove the next
@@ -94,8 +99,9 @@ int32_t    tsdbFreeMeta(STsdbMeta *pMeta);
 int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg);
 int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId);
 STable *tsdbIsValidTableToInsert(STsdbMeta *pMeta, STableId tableId);
-int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable);
+// int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable);
 STable *tsdbGetTableByUid(STsdbMeta *pMeta, int64_t uid);
+char *getTupleKey(const void * data);
 
 #ifdef __cplusplus
 }

src/vnode/tsdb/src/tsdbMain.c

@@ -311,10 +311,9 @@ int32_t tsdbTriggerCommit(tsdb_repo_t *repo) {
   // Loop to move pData to iData
   for (int i = 0; i < pRepo->config.maxTables; i++) {
     STable *pTable = pRepo->tsdbMeta->tables[i];
-    if (pTable != NULL) {
-      void *pData = pTable->content.pData;
-      pTable->content.pData = NULL;
-      pTable->iData = pData;
+    if (pTable != NULL && pTable->mem != NULL) {
+      pTable->imem = pTable->mem;
+      pTable->mem = NULL;
     }
   }
   // Loop to move mem to imem
@@ -669,7 +668,13 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
   int32_t level = 0;
   int32_t headSize = 0;
 
-  tSkipListRandNodeInfo(pTable->content.pData, &level, &headSize);
+  if (pTable->mem == NULL) {
+    pTable->mem = (SMemTable *)calloc(1, sizeof(SMemTable));
+    if (pTable->mem == NULL) return -1;
+    pTable->mem->pData = tSkipListCreate(5, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, getTupleKey);
+  }
+
+  tSkipListRandNodeInfo(pTable->mem->pData, &level, &headSize);
 
   // Copy row into the memory
   SSkipListNode *pNode = tsdbAllocFromCache(pRepo->tsdbCache, headSize + dataRowLen(row));
@@ -681,7 +686,7 @@ static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable
   dataRowCpy(SL_GET_NODE_DATA(pNode), row);
 
   // Insert the skiplist node into the data
-  tsdbInsertRowToTableImpl(pNode, pTable);
+  tSkipListPut(pTable->mem->pData, pNode);
 
   return 0;
 }
@@ -712,7 +717,7 @@ static void *tsdbCommitToFile(void *arg) {
   for (int i = 0; i < pRepo->config.maxTables; i++) {
     STable *pTable = pMeta->tables[i];
     if (pTable == NULL) continue;
-    SSkipListIterator *pIter = tSkipListCreateIter(pTable->iData);
+    SSkipListIterator *pIter = tSkipListCreateIter(pTable->imem->pData);
     while (tSkipListIterNext(pIter)) {
       SSkipListNode *node = tSkipListIterGet(pIter);
       SDataRow row = SL_GET_NODE_DATA(node);

src/vnode/tsdb/src/tsdbMeta.c

@@ -18,7 +18,6 @@ static int     tsdbAddTableIntoMap(STsdbMeta *pMeta, STable *pTable);
 static int     tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable);
 static int     tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable);
 static int     tsdbEstimateTableEncodeSize(STable *pTable);
-static char *  getTupleKey(const void *data);
 
 /**
  * Encode a TSDB table object as a binary content
@@ -102,12 +101,9 @@ int tsdbRestoreTable(void *pHandle, void *cont, int contLen) {
   if (pTable == NULL) return -1;
   
   if (pTable->type == TSDB_SUPER_TABLE) {
-    pTable->content.pIndex =
+    pTable->pIndex =
         tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 1, 0, getTupleKey);
-  } else {
-    pTable->content.pData = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, TSDB_DATA_TYPE_TIMESTAMP,
-                                            TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, getTupleKey);
-  }
+  } 
 
   tsdbAddTableToMeta(pMeta, pTable, false);
 
@@ -208,10 +204,10 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
       super->schema = tdDupSchema(pCfg->schema);
       super->tagSchema = tdDupSchema(pCfg->tagSchema);
       super->tagVal = tdDataRowDup(pCfg->tagValues);
-      super->content.pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 1,
+      super->pIndex = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 1,
                                                 0, getTupleKey);  // Allow duplicate key, no lock
 
-      if (super->content.pIndex == NULL) {
+      if (super->pIndex == NULL) {
         tdFreeSchema(super->schema);
         tdFreeSchema(super->tagSchema);
         tdFreeDataRow(super->tagVal);
@@ -223,7 +219,7 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
     }
   }
 
-  STable *table = (STable *)malloc(sizeof(STable));
+  STable *table = (STable *)calloc(1, sizeof(STable));
   if (table == NULL) {
     if (newSuper) tsdbFreeTable(super);
     return -1;
@@ -239,7 +235,6 @@ int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg) {
     table->superUid = -1;
     table->schema = tdDupSchema(pCfg->schema);
   }
-  table->content.pData = tSkipListCreate(TSDB_SUPER_TABLE_SL_LEVEL, TSDB_DATA_TYPE_TIMESTAMP, TYPE_BYTES[TSDB_DATA_TYPE_TIMESTAMP], 0, 0, getTupleKey);
 
   // Register to meta
   if (newSuper) tsdbAddTableToMeta(pMeta, super, true);
@@ -299,10 +294,10 @@ int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId) {
   return 0;
 }
 
-int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable) {
-  tSkipListPut(pTable->content.pData, pNode);
-  return 0;
-}
+// int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable) {
+//   tSkipListPut(pTable->mem->pData, pNode);
+//   return 0;
+// }
 
 static int tsdbFreeTable(STable *pTable) {
   // TODO: finish this function
@@ -314,10 +309,8 @@ static int tsdbFreeTable(STable *pTable) {
 
   // Free content
   if (TSDB_TABLE_IS_SUPER_TABLE(pTable)) {
-    tSkipListDestroy(pTable->content.pIndex);
-  } else {
-    tSkipListDestroy(pTable->content.pData);
-  }
+    tSkipListDestroy(pTable->pIndex);
+  } 
 
   free(pTable);
   return 0;
@@ -404,7 +397,7 @@ static int tsdbEstimateTableEncodeSize(STable *pTable) {
   return size;
 }
 
-static char *getTupleKey(const void * data) {
+char *getTupleKey(const void * data) {
   SDataRow row = (SDataRow)data;
 
   return dataRowAt(row, TD_DATA_ROW_HEAD_SIZE);

src/vnode/tsdb/tests/tsdbTests.cpp

@@ -40,7 +40,6 @@ TEST(TsdbTest, tableEncodeDecode) {
   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);
-  ASSERT_EQ(tTable->content.pData, nullptr);
 }
 
 TEST(TsdbTest, createRepo) {
@@ -72,7 +71,7 @@ TEST(TsdbTest, createRepo) {
   tsdbCreateTable(pRepo, &tCfg);
 
   // // 3. Loop to write some simple data
-  int nRows = 100;
+  int nRows = 1000;
   int rowsPerSubmit = 10;
   int64_t start_time = 1584081000000;