more

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

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;
+
+}

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
 }

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