add more code

src/vnode/common/inc/dataformat.h

@@ -16,6 +16,7 @@
 #define _TD_DATA_FORMAT_H_
 
 #include <stdint.h>
+#include <stdlib.h>
 
 #include "schema.h"
 
@@ -39,21 +40,28 @@ typedef void *SDataRow;
 #define dataRowTuple(r) ((char *)(r) + sizeof(int32_t))
 #define dataRowSetLen(r, l) (dataRowLen(r) = (l))
 #define dataRowIdx(r, i) ((char *)(r) + i)
+#define dataRowCpy(dst, r) memcpy((dst), (r), dataRowLen(r))
 
 SDataRow tdNewDataRow(int32_t bytes);
 SDataRow tdNewDdataFromSchema(SSchema *pSchema);
 void     tdFreeDataRow(SDataRow row);
 int32_t  tdAppendColVal(SDataRow row, void *value, SColumn *pCol, int32_t suffixOffset);
+void     tdDataRowCpy(void *dst, SDataRow row);
+void     tdDataRowReset(SDataRow row);
 
 /* Data rows definition, the format of it is like below:
- * +---------+---------+-----------------------+--------+-----------------------+
- * | int32_t | int32_t |                       |        |                       |
- * +---------+---------+-----------------------+--------+-----------------------+
- * |   len   |  nrows  |        SDataRow       |  ....  |        SDataRow       |
- * +---------+---------+-----------------------+--------+-----------------------+
+ * +---------+-----------------------+--------+-----------------------+
+ * | int32_t |                       |        |                       |
+ * +---------+-----------------------+--------+-----------------------+
+ * |   len   |        SDataRow       |  ....  |        SDataRow       |
+ * +---------+-----------------------+--------+-----------------------+
  */
 typedef void *SDataRows;
 
+#define dataRowsLen(rs) (*(int32_t *)(rs))
+#define dataRowsSetLen(rs, l) (dataRowsLen(rs) = (l))
+#define dataRowsInit(rs) dataRowsSetLen(rs, sizeof(int32_t))
+
 /* Data column definition
  * +---------+---------+-----------------------+
  * | int32_t | int32_t |                       |

src/vnode/common/src/dataformat.c

-#include <stdlib.h>
-
 #include "dataformat.h"
 
 /**
@@ -76,6 +74,15 @@ int32_t tdAppendColVal(SDataRow row, void *value, SColumn *pCol, int32_t suffixO
   return 0;
 }
 
+/**
+ * Copy a data row to a destination
+ * ASSUMPTIONS: dst has enough room for a copy of row
+ */
+void tdDataRowCpy(void *dst, SDataRow row) { memcpy(dst, row, dataRowLen(row)); }
+void tdDataRowReset(SDataRow row) { dataRowSetLen(row, sizeof(int32_t)); }
+
+// ------ Codes below should be refactored
+
 SDataRow tdSDataRowDup(SDataRow rdata) { return NULL; }
 void     tdFreeSDataRow(SDataRow rdata) {
   if (rdata == NULL) return;

src/vnode/tsdb/inc/tsdb.h

@@ -46,6 +46,7 @@ typedef struct {
 // Submit message for one table
 typedef struct {
   STableId tableId;
+  int32_t  padding;    // TODO just for padding here
   int32_t  sversion;   // data schema version
   int32_t  len;        // message length
   char     data[];

src/vnode/tsdb/src/tsdbMeta.c

@@ -184,7 +184,6 @@ static int tsdbFreeTable(STable *pTable) {
   // TODO: finish this function
   if (pTable->type == TSDB_STABLE) {
     tdFreeSDataRow(pTable->pTagVal);
-
   } else {
     tdFreeSchema(pTable->pSchema);
   }

src/vnode/tsdb/tests/tsdbTests.cpp

@@ -4,35 +4,37 @@
 #include "tsdb.h"
 #include "tsdbMeta.h"
 
-TEST(TsdbTest, createTable)  {
-    STsdbMeta *pMeta = tsdbCreateMeta(100);
-    ASSERT_NE(pMeta, nullptr);
-
-    STableCfg config;
-    config.tableId.tid = 0;
-    config.tableId.uid = 98868728187539L;
-    config.numOfCols = 5;
-    config.schema = tdNewSchema(config.numOfCols);
-    for (int i = 0; i < schemaNCols(config.schema); i++) {
-      SColumn *pCol = tdNewCol(TD_DATATYPE_BIGINT, i, 0);
-      tdColCpy(schemaColAt(config.schema, i), pCol);
-      tdFreeCol(pCol);
-    }
-    config.tagValues = nullptr;
+TEST(TsdbTest, createTable) {
+  STsdbMeta *pMeta = tsdbCreateMeta(100);
+  ASSERT_NE(pMeta, nullptr);
 
-    tsdbCreateTableImpl(pMeta, &config);
+  STableCfg config;
+  config.tableId.tid = 0;
+  config.tableId.uid = 98868728187539L;
+  config.numOfCols = 5;
+  config.schema = tdNewSchema(config.numOfCols);
+  for (int i = 0; i < schemaNCols(config.schema); i++) {
+    SColumn *pCol = tdNewCol(TD_DATATYPE_BIGINT, i, 0);
+    tdColCpy(schemaColAt(config.schema, i), pCol);
+    tdFreeCol(pCol);
+  }
+  config.tagValues = nullptr;
 
-    STable *pTable = tsdbGetTableByUid(pMeta, config.tableId.uid);
-    ASSERT_NE(pTable, nullptr);
+  tsdbCreateTableImpl(pMeta, &config);
+
+  STable *pTable = tsdbGetTableByUid(pMeta, config.tableId.uid);
+  ASSERT_NE(pTable, nullptr);
 }
 
-TEST(TsdbTest, createRepo)  {
+TEST(TsdbTest, createRepo) {
   STsdbCfg *pCfg = tsdbCreateDefaultCfg();
 
+  // Create a tsdb repository
   tsdb_repo_t *pRepo = tsdbCreateRepo("/root/mnt/test/vnode0", pCfg, NULL);
   ASSERT_NE(pRepo, nullptr);
   tsdbFreeCfg(pCfg);
 
+  // create a normal table in this repository
   STableCfg config;
   config.tableId.tid = 0;
   config.tableId.uid = 98868728187539L;
@@ -48,7 +50,44 @@ TEST(TsdbTest, createRepo)  {
   }
 
   tsdbCreateTable(pRepo, &config);
-  tdFreeSchema(config.schema);
+  // Write some data
 
+  int32_t size = sizeof(SSubmitMsg) + sizeof(SSubmitBlock) + tdMaxRowDataBytes(config.schema) * 10 + sizeof(int32_t);
+
+  SSubmitMsg *pMsg = (SSubmitMsg *)malloc(size);
+  pMsg->numOfTables = 1;  // TODO: use api
+
+  SSubmitBlock *pBlock = (SSubmitBlock *)pMsg->data;
+  pBlock->tableId = {.uid = 98868728187539L, .tid = 0};
+  pBlock->sversion = 0;
+  pBlock->len = sizeof(SSubmitBlock);
+
+  SDataRows rows = pBlock->data;
+  dataRowsInit(rows);
+
+  SDataRow row = tdNewDataRow(tdMaxRowDataBytes(config.schema));
+  int64_t ttime = 1583508800000;
+  void *pDst = pBlock->data;
+  for (int i = 0; i < 10; i++) {  // loop over rows
+    ttime += (10000 * i);
+    tdDataRowReset(row);
+    for (int j = 0; j < schemaNCols(config.schema); j++) {
+      if (j == 0) {  // set time stamp
+        tdAppendColVal(row, (void *)(&ttime), schemaColAt(config.schema, j), 24);
+      } else {       // set other fields
+        int val = 10;
+        tdAppendColVal(row, (void *)(&val), schemaColAt(config.schema, j), 24);
+      }
+    }
+
+    dataRowCpy((void *)pDst, row);
+    pDst += dataRowLen(row);
+  }
+
+  tsdbInsertData(pRepo, pMsg);
+
+  tdFreeDataRow(row);
+
+  tdFreeSchema(config.schema);
   tsdbDropRepo(pRepo);
 }
\ No newline at end of file