add more code

src/vnode/common/inc/dataformat.h

@@ -74,6 +74,7 @@ typedef struct {
 #define TD_DATAROW_DATA(pDataRow) ((pDataRow) + sizeof(int32_t))
 
 SDataRow tdSDataRowDup(SDataRow rdata);
+void tdSDataRowCpy(SDataRow src, void *dst);
 void tdFreeSDataRow(SDataRow rdata);
 
 // ---- operation on SDataRows

src/vnode/common/src/dataformat.c

@@ -52,3 +52,10 @@ int32_t tdRdataIterEnd(SDataRowsIter *pIter) {
   return pIter->rowCounter >= pIter->totalRows;
 
 }
+
+/**
+ * Copy it
+ */
+void tdSDataRowCpy(SDataRow src, void *dst) {
+  // TODO
+}

src/vnode/tsdb/inc/tsdbCache.h

@@ -54,8 +54,8 @@ typedef struct STSDBCache {
 #define TSDB_PREV_CACHE_BLOCK(pBlock) ((pBlock)->prev)
 
 STsdbCache *tsdbCreateCache(int32_t numOfBlocks);
-int32_t tsdbFreeCache(STsdbCache *pCache);
-void *tsdbAllocFromCache(STsdbCache *pCache, int64_t bytes);
+int32_t     tsdbFreeCache(STsdbCache *pCache);
+void *      tsdbAllocFromCache(STsdbCache *pCache, int64_t bytes);
 
 #ifdef __cplusplus
 }

src/vnode/tsdb/inc/tsdbMeta.h

@@ -19,6 +19,7 @@
 
 #include "tsdb.h"
 #include "dataformat.h"
+#include "tskiplist.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -114,6 +115,7 @@ STsdbMeta *tsdbOpenMeta(char *tsdbDir);
 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);
 
 #ifdef __cplusplus
 }

src/vnode/tsdb/src/tsdbMain.c

@@ -17,6 +17,7 @@
 #include "tsdbFile.h"
 #include "tsdbMeta.h"
 #include "tutil.h"
+#include "tskiplist.h"
 
 #define TSDB_DEFAULT_PRECISION TSDB_PRECISION_MILLI  // default precision
 #define IS_VALID_PRECISION(precision) (((precision) >= TSDB_PRECISION_MILLI) && ((precision) <= TSDB_PRECISION_NANO))
@@ -403,7 +404,25 @@ static int tsdbRecoverRepo(int fd, STsdbCfg *pCfg) {
   return 0;
 }
 
-static FORCE_INLINE int32_t tdInsertRowToTable(SDataRow row, STable *pTable) { return 0; }
+static int32_t tdInsertRowToTable(STsdbRepo *pRepo, SDataRow row, STable *pTable) {
+  // TODO
+  int32_t level = 0;
+  int32_t headSize = 0;
+
+  // Copy row into the memory
+  SSkipListNode *pNode = tsdbAllocFromCache(pRepo->tsdbCache, headSize + TD_DATAROW_LEN(row));
+  if (pNode == NULL) {
+    // TODO: deal with allocate failure
+  }
+
+  pNode->level = level;
+  tdSDataRowCpy(row, SL_GET_NODE_DATA(pNode));
+
+  // Insert the skiplist node into the data
+  tsdbInsertRowToTableImpl(pNode, pTable);
+
+  return 0;
+}
 
 static int32_t tsdbInsertDataToTable(tsdb_repo_t *repo, SSubmitBlock *pBlock) {
   STsdbRepo *pRepo = (STsdbRepo *)repo;
@@ -418,7 +437,7 @@ static int32_t tsdbInsertDataToTable(tsdb_repo_t *repo, SSubmitBlock *pBlock) {
 
   tdInitSDataRowsIter(rows, pIter);
   while (!tdRdataIterEnd(pIter)) {
-    if (tdInsertRowToTable(pIter->row, pTable) < 0) {
+    if (tdInsertRowToTable(pRepo, pIter->row, pTable) < 0) {
       // TODO: deal with the error here
     }
 

src/vnode/tsdb/src/tsdbMeta.c

@@ -175,6 +175,11 @@ int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId) {
   }
 }
 
+int32_t tsdbInsertRowToTableImpl(SSkipListNode *pNode, STable *pTable) {
+  tSkipListPut(pTable->content.pData, pNode);
+  return 0;
+}
+
 static int tsdbFreeTable(STable *pTable) {
   // TODO: finish this function
   if (pTable->type == TSDB_STABLE) {