fix tsdb insert

source/dnode/vnode/src/inc/tsdb.h

@@ -38,7 +38,7 @@ typedef struct STable        STable;
 
 int  tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable);
 void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable);
-int  tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitReq *pMsg, SSubmitRsp *pRsp);
+int  tsdbInsertTableData(STsdb *pTsdb, SSubmitBlk *pBlock, int32_t *pAffectedRows);
 int  tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey, int maxRowsToRead, SDataCols *pCols,
                            TKEY *filterKeys, int nFilterKeys, bool keepDup, SMergeInfo *pMergeInfo);
 
@@ -62,7 +62,7 @@ struct STable {
 
 int     tsdbOpen(SVnode *pVnode, STsdb **ppTsdb);
 int     tsdbClose(STsdb *pTsdb);
-int     tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp);
+int     tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *pRsp);
 int     tsdbPrepareCommit(STsdb *pTsdb);
 int     tsdbCommit(STsdb *pTsdb);
 int32_t tsdbInitSma(STsdb *pTsdb);

source/dnode/vnode/src/tsdb/tsdbMemTable.c

@@ -15,8 +15,6 @@
 
 #include "vnodeInt.h"
 
-static int      tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg);
-static int      tsdbMemTableInsertTbData(STsdb *pRepo, SSubmitBlk *pBlock, int32_t *pAffectedRows);
 static STbData *tsdbNewTbData(tb_uid_t uid);
 static void     tsdbFreeTbData(STbData *pTbData);
 static char    *tsdbGetTsTupleKey(const void *data);
@@ -57,6 +55,7 @@ int tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable) {
     return -1;
   }
 
+  *ppMemTable = pMemTable;
   return 0;
 }
 
@@ -68,37 +67,6 @@ void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable) {
   }
 }
 
-int tsdbMemTableInsert(STsdb *pTsdb, STsdbMemTable *pMemTable, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
-  SSubmitBlk    *pBlock = NULL;
-  SSubmitMsgIter msgIter = {0};
-  int32_t        affectedrows = 0, numOfRows = 0;
-
-  if (tsdbScanAndConvertSubmitMsg(pTsdb, pMsg) < 0) {
-    if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-      tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pTsdb), tstrerror(terrno));
-    }
-    return -1;
-  }
-
-  tInitSubmitMsgIter(pMsg, &msgIter);
-  while (true) {
-    tGetSubmitMsgNext(&msgIter, &pBlock);
-    if (pBlock == NULL) break;
-    if (tsdbMemTableInsertTbData(pTsdb, pBlock, &affectedrows) < 0) {
-      return -1;
-    }
-
-    numOfRows += pBlock->numOfRows;
-  }
-
-  if (pRsp != NULL) {
-    pRsp->affectedRows = affectedrows;
-    pRsp->numOfRows = numOfRows;
-  }
-
-  return 0;
-}
-
 /**
  * This is an important function to load data or try to load data from memory skiplist iterator.
  *
@@ -250,78 +218,7 @@ int32_t tdScanAndConvertSubmitMsg(SSubmitReq *pMsg) {
   return 0;
 }
 
-static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
-  ASSERT(pMsg != NULL);
-  // STsdbMeta *    pMeta = pTsdb->tsdbMeta;
-  SSubmitMsgIter msgIter = {0};
-  SSubmitBlk    *pBlock = NULL;
-  SSubmitBlkIter blkIter = {0};
-  STSRow        *row = NULL;
-  TSKEY          now = taosGetTimestamp(pTsdb->config.precision);
-  TSKEY          minKey = now - tsTickPerDay[pTsdb->config.precision] * pTsdb->config.keep2;
-  TSKEY          maxKey = now + tsTickPerDay[pTsdb->config.precision] * pTsdb->config.days;
-
-  terrno = TSDB_CODE_SUCCESS;
-  pMsg->length = htonl(pMsg->length);
-  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
-
-  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
-  while (true) {
-    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
-    if (pBlock == NULL) break;
-
-    pBlock->uid = htobe64(pBlock->uid);
-    pBlock->suid = htobe64(pBlock->suid);
-    pBlock->sversion = htonl(pBlock->sversion);
-    pBlock->dataLen = htonl(pBlock->dataLen);
-    pBlock->schemaLen = htonl(pBlock->schemaLen);
-    pBlock->numOfRows = htons(pBlock->numOfRows);
-
-#if 0
-    if (pBlock->tid <= 0 || pBlock->tid >= pMeta->maxTables) {
-      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pTsdb), pBlock->uid,
-                pBlock->tid);
-      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-      return -1;
-    }
-
-    STable *pTable = pMeta->tables[pBlock->tid];
-    if (pTable == NULL || TABLE_UID(pTable) != pBlock->uid) {
-      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pTsdb), pBlock->uid,
-                pBlock->tid);
-      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-      return -1;
-    }
-
-    if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
-      tsdbError("vgId:%d invalid action trying to insert a super table %s", REPO_ID(pTsdb), TABLE_CHAR_NAME(pTable));
-      terrno = TSDB_CODE_TDB_INVALID_ACTION;
-      return -1;
-    }
-
-    // Check schema version and update schema if needed
-    if (tsdbCheckTableSchema(pTsdb, pBlock, pTable) < 0) {
-      if (terrno == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-        continue;
-      } else {
-        return -1;
-      }
-    }
-
-    tsdbInitSubmitBlkIter(pBlock, &blkIter);
-    while ((row = tsdbGetSubmitBlkNext(&blkIter)) != NULL) {
-      if (tsdbCheckRowRange(pTsdb, pTable, row, minKey, maxKey, now) < 0) {
-        return -1;
-      }
-    }
-#endif
-  }
-
-  if (terrno != TSDB_CODE_SUCCESS) return -1;
-  return 0;
-}
-
-static int tsdbMemTableInsertTbData(STsdb *pTsdb, SSubmitBlk *pBlock, int32_t *pAffectedRows) {
+int tsdbInsertTableData(STsdb *pTsdb, SSubmitBlk *pBlock, int32_t *pAffectedRows) {
   // STsdbMeta       *pMeta = pRepo->tsdbMeta;
   // int32_t          points = 0;
   // STable          *pTable = NULL;
@@ -332,11 +229,9 @@ static int tsdbMemTableInsertTbData(STsdb *pTsdb, SSubmitBlk *pBlock, int32_t *p
   STSRow        *row;
   TSKEY          keyMin;
   TSKEY          keyMax;
+  SSubmitBlk    *pBlkCopy;
 
-  // SMemTable       *pMemTable = NULL;
-  // STableData      *pTableData = NULL;
-  // STsdbCfg        *pCfg = &(pRepo->config);
-
+  // create container is nedd
   tptr = taosHashGet(pMemTable->pHashIdx, &(pBlock->uid), sizeof(pBlock->uid));
   if (tptr == NULL) {
     pTbData = tsdbNewTbData(pBlock->uid);
@@ -353,7 +248,11 @@ static int tsdbMemTableInsertTbData(STsdb *pTsdb, SSubmitBlk *pBlock, int32_t *p
     pTbData = *(STbData **)tptr;
   }
 
-  tInitSubmitBlkIter(pBlock, &blkIter);
+  // copy data to buffer pool
+  pBlkCopy = (SSubmitBlk *)vnodeBufPoolMalloc(pTsdb->mem->pPool, pBlock->dataLen + sizeof(*pBlock));
+  memcpy(pBlkCopy, pBlock, pBlock->dataLen + sizeof(*pBlock));
+
+  tInitSubmitBlkIter(pBlkCopy, &blkIter);
   if (blkIter.row == NULL) return 0;
   keyMin = TD_ROW_KEY(blkIter.row);
 
@@ -372,31 +271,6 @@ static int tsdbMemTableInsertTbData(STsdb *pTsdb, SSubmitBlk *pBlock, int32_t *p
 
   (*pAffectedRows) += pBlock->numOfRows;
 
-  // STSRow* lastRow = NULL;
-  // int64_t osize = SL_SIZE(pTableData->pData);
-  // tsdbSetupSkipListHookFns(pTableData->pData, pRepo, pTable, &points, &lastRow);
-  // tSkipListPutBatchByIter(pTableData->pData, &blkIter, (iter_next_fn_t)tsdbGetSubmitBlkNext);
-  // int64_t dsize = SL_SIZE(pTableData->pData) - osize;
-  // (*pAffectedRows) += points;
-
-  // if(lastRow != NULL) {
-  //   TSKEY lastRowKey = TD_ROW_KEY(lastRow);
-  //   if (pMemTable->keyFirst > firstRowKey) pMemTable->keyFirst = firstRowKey;
-  //   pMemTable->numOfRows += dsize;
-
-  //   if (pTableData->keyFirst > firstRowKey) pTableData->keyFirst = firstRowKey;
-  //   pTableData->numOfRows += dsize;
-  //   if (pMemTable->keyLast < lastRowKey) pMemTable->keyLast = lastRowKey;
-  //   if (pTableData->keyLast < lastRowKey) pTableData->keyLast = lastRowKey;
-  //   if (tsdbUpdateTableLatestInfo(pRepo, pTable, lastRow) < 0) {
-  //     return -1;
-  //   }
-  // }
-
-  // STSchema *pSchema = tsdbGetTableSchemaByVersion(pTable, pBlock->sversion, -1);
-  // pRepo->stat.pointsWritten += points * schemaNCols(pSchema);
-  // pRepo->stat.totalStorage += points * schemaVLen(pSchema);
-
   return 0;
 }
 
@@ -521,7 +395,6 @@ static int          tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables);
 static int          tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, STSRow* row);
 static int          tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter);
 static STSRow*      tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter);
-static int          tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitReq *pMsg);
 static int          tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, int32_t *affectedrows);
 static int          tsdbInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter);
 static int          tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock);

source/dnode/vnode/src/tsdb/tsdbWrite.c

@@ -15,21 +15,111 @@
 
 #include "vnodeInt.h"
 
-/**
- * @brief insert TS data
- *
- * @param pTsdb
- * @param pMsg
- * @param pRsp
- * @return int
- */
-int tsdbInsertData(STsdb *pTsdb, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
-  // Check if mem is there. If not, create one.
-  // if (pTsdb->mem == NULL) {
-  //   pTsdb->mem = tsdbMemTableCreate(pTsdb);
-  //   if (pTsdb->mem == NULL) {
-  //     return -1;
-  //   }
-  // }
-  return tsdbMemTableInsert(pTsdb, pTsdb->mem, pMsg, pRsp);
+static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg);
+
+int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
+  SSubmitMsgIter msgIter = {0};
+  SSubmitBlk    *pBlock = NULL;
+  int32_t        affectedrows = 0;
+  int32_t        numOfRows = 0;
+
+  ASSERT(pTsdb->mem != NULL);
+
+  // scan and convert
+  if (tsdbScanAndConvertSubmitMsg(pTsdb, pMsg) < 0) {
+    if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
+      tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pTsdb), tstrerror(terrno));
+    }
+    return -1;
+  }
+
+  // loop to insert
+  tInitSubmitMsgIter(pMsg, &msgIter);
+  while (true) {
+    tGetSubmitMsgNext(&msgIter, &pBlock);
+    if (pBlock == NULL) break;
+    if (tsdbInsertTableData(pTsdb, pBlock, &affectedrows) < 0) {
+      return -1;
+    }
+
+    numOfRows += pBlock->numOfRows;
+  }
+
+  if (pRsp != NULL) {
+    pRsp->affectedRows = affectedrows;
+    pRsp->numOfRows = numOfRows;
+  }
+
+  return 0;
+}
+
+static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
+  ASSERT(pMsg != NULL);
+  // STsdbMeta *    pMeta = pTsdb->tsdbMeta;
+  SSubmitMsgIter msgIter = {0};
+  SSubmitBlk    *pBlock = NULL;
+  SSubmitBlkIter blkIter = {0};
+  STSRow        *row = NULL;
+  TSKEY          now = taosGetTimestamp(pTsdb->config.precision);
+  TSKEY          minKey = now - tsTickPerDay[pTsdb->config.precision] * pTsdb->config.keep2;
+  TSKEY          maxKey = now + tsTickPerDay[pTsdb->config.precision] * pTsdb->config.days;
+
+  terrno = TSDB_CODE_SUCCESS;
+  pMsg->length = htonl(pMsg->length);
+  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
+
+  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
+  while (true) {
+    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
+    if (pBlock == NULL) break;
+
+    pBlock->uid = htobe64(pBlock->uid);
+    pBlock->suid = htobe64(pBlock->suid);
+    pBlock->sversion = htonl(pBlock->sversion);
+    pBlock->dataLen = htonl(pBlock->dataLen);
+    pBlock->schemaLen = htonl(pBlock->schemaLen);
+    pBlock->numOfRows = htons(pBlock->numOfRows);
+
+#if 0
+    if (pBlock->tid <= 0 || pBlock->tid >= pMeta->maxTables) {
+      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pTsdb), pBlock->uid,
+                pBlock->tid);
+      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
+      return -1;
+    }
+
+    STable *pTable = pMeta->tables[pBlock->tid];
+    if (pTable == NULL || TABLE_UID(pTable) != pBlock->uid) {
+      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pTsdb), pBlock->uid,
+                pBlock->tid);
+      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
+      return -1;
+    }
+
+    if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
+      tsdbError("vgId:%d invalid action trying to insert a super table %s", REPO_ID(pTsdb), TABLE_CHAR_NAME(pTable));
+      terrno = TSDB_CODE_TDB_INVALID_ACTION;
+      return -1;
+    }
+
+    // Check schema version and update schema if needed
+    if (tsdbCheckTableSchema(pTsdb, pBlock, pTable) < 0) {
+      if (terrno == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
+        continue;
+      } else {
+        return -1;
+      }
+    }
+
+    tsdbInitSubmitBlkIter(pBlock, &blkIter);
+    while ((row = tsdbGetSubmitBlkNext(&blkIter)) != NULL) {
+      if (tsdbCheckRowRange(pTsdb, pTable, row, minKey, maxKey, now) < 0) {
+        return -1;
+      }
+    }
+#endif
+  }
+
+  if (terrno != TSDB_CODE_SUCCESS) return -1;
+  return 0;
 }
\ No newline at end of file

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -62,6 +62,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
   metaRsp.tuid = mer1.me.uid;
 
   if (mer1.me.type == TSDB_SUPER_TABLE) {
+    strcpy(metaRsp.stbName, mer1.me.name);
     schema = mer1.me.stbEntry.schema;
     schemaTag = mer1.me.stbEntry.schemaTag;
     metaRsp.suid = mer1.me.uid;
@@ -69,6 +70,7 @@ int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
     metaReaderInit(&mer2, pVnode, 0);
     if (metaGetTableEntryByUid(&mer2, mer1.me.ctbEntry.suid) < 0) goto _exit;
 
+    strcpy(metaRsp.stbName, mer2.me.name);
     metaRsp.suid = mer2.me.uid;
     schema = mer2.me.stbEntry.schema;
     schemaTag = mer2.me.stbEntry.schemaTag;

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -21,7 +21,7 @@ static int vnodeProcessDropStbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcM
 static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp);
 static int vnodeProcessAlterTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcMsg *pRsp);
 static int vnodeProcessDropTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcMsg *pRsp);
-static int vnodeProcessSubmitReq(SVnode *pVnode, SSubmitReq *pSubmitReq, SRpcMsg *pRsp);
+static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
 
 int vnodePreprocessWriteReqs(SVnode *pVnode, SArray *pMsgs, int64_t *version) {
   SNodeMsg *pMsg;
@@ -92,8 +92,7 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
     } break;
     /* TSDB */
     case TDMT_VND_SUBMIT:
-      pRsp->msgType = TDMT_VND_SUBMIT_RSP;
-      vnodeProcessSubmitReq(pVnode, ptr, pRsp);
+      if (vnodeProcessSubmitReq(pVnode, version, pMsg->pCont, pMsg->contLen, pRsp) < 0) goto _err;
       break;
     /* TQ */
     case TDMT_VND_MQ_VG_CHANGE:
@@ -352,13 +351,14 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, void *pReq, int32_t len, SRpcMs
   return 0;
 }
 
-static int vnodeProcessSubmitReq(SVnode *pVnode, SSubmitReq *pSubmitReq, SRpcMsg *pRsp) {
-  SSubmitRsp rsp = {0};
+static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
+  SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;
+  SSubmitRsp  rsp = {0};
 
   pRsp->code = 0;
 
   // handle the request
-  if (tsdbInsertData(pVnode->pTsdb, pSubmitReq, &rsp) < 0) {
+  if (tsdbInsertData(pVnode->pTsdb, version, pSubmitReq, &rsp) < 0) {
     pRsp->code = terrno;
     return -1;
   }