auto create table

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

@@ -40,7 +40,6 @@ typedef struct STable        STable;
 
 int  tsdbMemTableCreate(STsdb *pTsdb, STsdbMemTable **ppMemTable);
 void tsdbMemTableDestroy(STsdb *pTsdb, STsdbMemTable *pMemTable);
-int  tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, 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);
 

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

@@ -100,6 +100,7 @@ int32_t      tsdbUpdateSmaWindow(STsdb* pTsdb, SSubmitReq* pMsg, int64_t version
 int32_t      tsdbCreateTSma(STsdb* pTsdb, char* pMsg);
 int32_t      tsdbInsertTSmaData(STsdb* pTsdb, int64_t indexUid, const char* msg);
 int          tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
+int          tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, int32_t* pAffectedRows);
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
                              uint64_t taskId);
 tsdbReaderT  tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
@@ -125,7 +126,6 @@ int32_t tsdbUpdateTbUidList(STsdb* pTsdb, STbUidStore* pUidStore);
 void    tsdbUidStoreDestory(STbUidStore* pStore);
 void*   tsdbUidStoreFree(STbUidStore* pStore);
 int32_t tsdbTriggerRSma(STsdb* pTsdb, void* pMsg, int32_t inputType);
-int32_t tsdbProcessSubmitReq(STsdb* pTsdb, int64_t version, void* pReq);
 
 typedef struct {
   int8_t  streamType;  // sma or other

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

@@ -69,10 +69,6 @@ static void tsdbDestroyCommitIters(SCommitH *pCommith);
 static int  tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
 static void tsdbResetCommitFile(SCommitH *pCommith);
 static int  tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid);
-// static int  tsdbCommitMeta(STsdbRepo *pRepo);
-// static int  tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen, bool compact);
-// static int  tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid);
-// static int  tsdbCompactMetaFile(STsdbRepo *pRepo, STsdbFS *pfs, SMFile *pMFile);
 static int  tsdbCommitToTable(SCommitH *pCommith, int tid);
 static int  tsdbSetCommitTable(SCommitH *pCommith, STable *pTable);
 static int  tsdbComparKeyBlock(const void *arg1, const void *arg2);

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

@@ -191,9 +191,6 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
 }
 
 int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, int32_t *pAffectedRows) {
-  // STsdbMeta       *pMeta = pRepo->tsdbMeta;
-  // int32_t          points = 0;
-  // STable          *pTable = NULL;
   SSubmitBlkIter blkIter = {0};
   STsdbMemTable *pMemTable = pTsdb->mem;
   void          *tptr;
@@ -221,8 +218,9 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
   }
 
   // copy data to buffer pool
-  pBlkCopy = (SSubmitBlk *)vnodeBufPoolMalloc(pTsdb->mem->pPool, pMsgIter->dataLen + sizeof(*pBlock));
-  memcpy(pBlkCopy, pBlock, pMsgIter->dataLen + sizeof(*pBlock));
+  int32_t tlen = pMsgIter->dataLen + pMsgIter->schemaLen + sizeof(*pBlock);
+  pBlkCopy = (SSubmitBlk *)vnodeBufPoolMalloc(pTsdb->mem->pPool, tlen);
+  memcpy(pBlkCopy, pBlock, tlen);
 
   tInitSubmitBlkIter(pMsgIter, pBlkCopy, &blkIter);
   if (blkIter.row == NULL) return 0;
@@ -241,7 +239,7 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
   if (pMemTable->keyMin > keyMin) pMemTable->keyMin = keyMin;
   if (pMemTable->keyMax < keyMax) pMemTable->keyMax = keyMax;
 
-  (*pAffectedRows) += pMsgIter->numOfRows;
+  (*pAffectedRows) = pMsgIter->numOfRows;
 
   return 0;
 }

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

@@ -1962,6 +1962,21 @@ int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pStore) {
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t tsdbProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
+  if (!pReq) {
+    terrno = TSDB_CODE_INVALID_PTR;
+    return TSDB_CODE_FAILED;
+  }
+
+  SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;
+
+  if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) {
+    return TSDB_CODE_FAILED;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t tsdbFetchSubmitReqSuids(SSubmitReq *pMsg, STbUidStore *pStore) {
   ASSERT(pMsg != NULL);
   SSubmitMsgIter msgIter = {0};

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

@@ -488,20 +488,55 @@ _exit:
 }
 
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
-  SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;
-  SSubmitRsp  rsp = {0};
+  SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
+  SSubmitMsgIter msgIter = {0};
+  SSubmitBlk    *pBlock;
+  SSubmitRsp     rsp = {0};
+  SVCreateTbReq  createTbReq = {0};
+  SCoder         coder = {0};
+  int32_t        nRows;
 
   pRsp->code = 0;
 
   // handle the request
-  if (tsdbInsertData(pVnode->pTsdb, version, pSubmitReq, &rsp) < 0) {
-    pRsp->code = terrno;
-    return -1;
+  if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
+    pRsp->code = TSDB_CODE_INVALID_MSG;
+    goto _exit;
   }
 
-  // pRsp->msgType = TDMT_VND_SUBMIT_RSP;
-  // vnodeProcessSubmitReq(pVnode, ptr, pRsp);
+  for (;;) {
+    tGetSubmitMsgNext(&msgIter, &pBlock);
+    if (pBlock == NULL) break;
+
+    // create table for auto create table mode
+    if (msgIter.schemaLen > 0) {
+      tCoderInit(&coder, TD_LITTLE_ENDIAN, pBlock->data, msgIter.schemaLen, TD_DECODER);
+      if (tDecodeSVCreateTbReq(&coder, &createTbReq) < 0) {
+        pRsp->code = TSDB_CODE_INVALID_MSG;
+        tCoderClear(&coder);
+        goto _exit;
+      }
+
+      if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
+        if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
+          pRsp->code = terrno;
+          tCoderClear(&coder);
+          goto _exit;
+        }
+      }
+
+      tCoderClear(&coder);
+    }
+
+    if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &nRows) < 0) {
+      pRsp->code = terrno;
+      goto _exit;
+    }
+
+    rsp.numOfRows += nRows;
+  }
 
+_exit:
   // encode the response (TODO)
   pRsp->pCont = rpcMallocCont(sizeof(SSubmitRsp));
   memcpy(pRsp->pCont, &rsp, sizeof(rsp));
@@ -511,18 +546,3 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 
   return 0;
 }
-
-int32_t tsdbProcessSubmitReq(STsdb *pTsdb, int64_t version, void *pReq) {
-  if (!pReq) {
-    terrno = TSDB_CODE_INVALID_PTR;
-    return TSDB_CODE_FAILED;
-  }
-
-  SSubmitReq *pSubmitReq = (SSubmitReq *)pReq;
-
-  if (tsdbInsertData(pTsdb, version, pSubmitReq, NULL) < 0) {
-    return TSDB_CODE_FAILED;
-  }
-
-  return TSDB_CODE_SUCCESS;
-}