From 8367106266d81ad9122a5b42348dc0496d62c61b Mon Sep 17 00:00:00 2001
From: hzcheng <hzcheng@taosdata.com>
Date: Wed, 4 Mar 2020 13:28:19 +0000
Subject: [PATCH] more

---
 src/vnode/common/inc/dataformat.h | 15 +++++++++++++--
 src/vnode/common/src/dataformat.c | 18 +++++++++++++++++-
 src/vnode/tsdb/inc/tsdbMeta.h     |  2 +-
 src/vnode/tsdb/src/tsdbMeta.c     | 27 ++++++++++++++++++++-------
 4 files changed, 51 insertions(+), 11 deletions(-)

diff --git a/src/vnode/common/inc/dataformat.h b/src/vnode/common/inc/dataformat.h
index db2ab72a83..45a76bb328 100644
--- a/src/vnode/common/inc/dataformat.h
+++ b/src/vnode/common/inc/dataformat.h
@@ -60,6 +60,12 @@ typedef char * SDataCol;
  */
 typedef char * SDataCols;
 
+typedef struct {
+    int32_t rowCounter;
+    int32_t totalRows;
+    SDataRow row;
+} SDataRowsIter;
+
 // ----------------- Data column structure
 
 // ---- operation on SDataRow;
@@ -72,8 +78,8 @@ void tdFreeSDataRow(SDataRow rdata);
 
 // ---- operation on SDataRows
 #define TD_DATAROWS_LEN(pDataRows) (*(int32_t *)(pDataRows))
-#define TD_DATAROWS_ROWS(pDataRows) (*(int32_t *)(pDataRows + sizeof(int32_t)))
-#define TD_NEXT_DATAROW(pDataRow)  ((pDataRow) + TD_DATAROW_LEN(pDataRow))
+#define TD_DATAROWS_ROWS(pDataRows) (*(int32_t *)((pDataRows) + sizeof(int32_t)))
+#define TD_DATAROWS_DATA(pDataRows) (SDataRow)((pDataRows) + 2 * sizeof(int32_t))
 
 // ---- operation on SDataCol
 #define TD_DATACOL_LEN(pDataCol) (*(int32_t *)(pDataCol))
@@ -83,6 +89,11 @@ void tdFreeSDataRow(SDataRow rdata);
 #define TD_DATACOLS_LEN(pDataCols) (*(int32_t *)(pDataCols))
 #define TD_DATACOLS_NPOINTS(pDataCols) (*(int32_t *)(pDataCols + sizeof(int32_t)))
 
+// ---- operation on SDataRowIter
+int32_t tdInitSDataRowsIter(SDataRows rows, SDataRowsIter *pIter);
+int32_t tdRdataIterEnd(SDataRowsIter *pIter);
+void    tdRdataIterNext(SDataRowsIter *pIter);
+
 // ----
 /**
  * Get the maximum
diff --git a/src/vnode/common/src/dataformat.c b/src/vnode/common/src/dataformat.c
index 23351e9365..79a58ac45a 100644
--- a/src/vnode/common/src/dataformat.c
+++ b/src/vnode/common/src/dataformat.c
@@ -35,4 +35,20 @@ SDataRow tdSDataRowDup(SDataRow rdata) { return NULL; }
 void     tdFreeSDataRow(SDataRow rdata) {
   if (rdata == NULL) return;
   free(rdata);
-}
\ No newline at end of file
+}
+
+int32_t tdInitSDataRowsIter(SDataRows rows, SDataRowsIter *pIter) {
+  pIter->totalRows = TD_DATAROWS_ROWS(rows);
+  pIter->rowCounter = 1;
+  pIter->row = TD_DATAROWS_DATA(rows);
+}
+
+void tdRdataIterNext(SDataRowsIter *pIter) {
+  pIter->rowCounter++;
+  pIter->row = pIter->row + TD_DATAROW_LEN(pIter->row);
+}
+
+int32_t tdRdataIterEnd(SDataRowsIter *pIter) {
+  return pIter->rowCounter >= pIter->totalRows;
+
+}
diff --git a/src/vnode/tsdb/inc/tsdbMeta.h b/src/vnode/tsdb/inc/tsdbMeta.h
index 240716121a..36c2dbd6f5 100644
--- a/src/vnode/tsdb/inc/tsdbMeta.h
+++ b/src/vnode/tsdb/inc/tsdbMeta.h
@@ -114,7 +114,7 @@ STsdbMeta *tsdbOpenMeta(char *tsdbDir);
 int32_t tsdbCreateTableImpl(STsdbMeta *pMeta, STableCfg *pCfg);
 int32_t tsdbDropTableImpl(STsdbMeta *pMeta, STableId tableId);
 
-int32_t tsdbInsertDataImpl(STsdbMeta *pMeta, STableId tableId, char *pData);
+int32_t tsdbInsertDataImpl(STsdbMeta *pMeta, STableId tableId, SDataRows rows);
 
 #ifdef __cplusplus
 }
diff --git a/src/vnode/tsdb/src/tsdbMeta.c b/src/vnode/tsdb/src/tsdbMeta.c
index 00a9469cc6..46bea01fbe 100644
--- a/src/vnode/tsdb/src/tsdbMeta.c
+++ b/src/vnode/tsdb/src/tsdbMeta.c
@@ -16,6 +16,7 @@ static int     tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable);
 static int     tsdbAddTableIntoMap(STsdbMeta *pMeta, STable *pTable);
 static int     tsdbAddTableIntoIndex(STsdbMeta *pMeta, STable *pTable);
 static int     tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable);
+static int tsdbInsertRowToTable(STable *pTable, SDataRow row);
 
 STsdbMeta *tsdbCreateMeta(int32_t maxTables) {
   STsdbMeta *pMeta = (STsdbMeta *)malloc(sizeof(STsdbMeta));
@@ -136,16 +137,23 @@ STsdbMeta *tsdbOpenMeta(char *tsdbDir) {
   return pMeta;
 }
 
-int32_t tsdbInsertDataImpl(STsdbMeta *pMeta, STableId tableId, char *pData) {
-  STable *pTable = pMeta->tables[tableId.tid];
+int32_t tsdbInsertDataImpl(STsdbMeta *pMeta, STableId tableId, SDataRows rows) {
+  STable *pTable = tsdbGetTableByUid(pMeta, tableId.uid);
   if (pTable == NULL) {
-    // TODO: deal with the error here
-    return 0;
+    return -1;
   }
 
-  if (pTable->tableId.uid != tableId.uid) {
-    // TODO: deal with the error here
-    return 0;
+  if (TSDB_TABLE_IS_SUPER_TABLE(pTable)) return -1;
+  if (pTable->tableId.tid != tableId.tid) return -1;
+
+  // Loop to write each row
+  SDataRowsIter sdataIter;
+  tdInitSDataRowsIter(rows, &sdataIter);
+  while (!tdRdataIterEnd(&sdataIter)) {
+    // Insert the row to it
+    tsdbInsertRowToTable(pTable, sdataIter.row);
+
+    tdRdataIterNext(&sdataIter);
   }
 
   return 0;
@@ -247,4 +255,9 @@ static int tsdbRemoveTableFromIndex(STsdbMeta *pMeta, STable *pTable) {
   assert(pTable->type == TSDB_STABLE);
   // TODO
   return 0;
+}
+
+static int tsdbInsertRowToTable(STable *pTable, SDataRow row) {
+  // TODO
+  return 0;
 }
\ No newline at end of file
-- 
GitLab