Merge remote-tracking branch 'origin/enh/rocksdbSstate' into enh/rocksdbSstate

include/libs/stream/tstream.h

@@ -239,6 +239,7 @@ typedef struct {
   void*           vnode;  // not available to encoder and decoder
   FTbSink*        tbSinkFunc;
   STSchema*       pTSchema;
+  SSHashObj*      pTblInfo;
 } STaskSinkTb;
 
 typedef void FSmaSink(void* vnode, int64_t smaId, const SArray* data);

source/dnode/vnode/src/tq/tq.c

@@ -572,6 +572,10 @@ end:
   return ret;
 }
 
+void freePtr(void *ptr) {
+  taosMemoryFree(*(void**)ptr);
+}
+
 int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) {
   int32_t vgId = TD_VID(pTq->pVnode);
   pTask->id.idStr = createStreamTaskIdStr(pTask->id.streamId, pTask->id.taskId);
@@ -646,6 +650,8 @@ int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask, int64_t ver) {
     if (pTask->tbSink.pTSchema == NULL) {
       return -1;
     }
+    pTask->tbSink.pTblInfo = tSimpleHashInit(10240, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
+    tSimpleHashSetFreeFp(pTask->tbSink.pTblInfo, freePtr);
   }
 
   if (pTask->taskLevel == TASK_LEVEL__SOURCE) {

source/dnode/vnode/src/tq/tqSink.c

@@ -17,6 +17,13 @@
 #include "tmsg.h"
 #include "tq.h"
 
+#define MAX_CATCH_NUM 10240
+
+typedef struct STblInfo {
+  uint64_t uid;
+  char     tbName[TSDB_TABLE_NAME_LEN];
+} STblInfo;
+
 int32_t tqBuildDeleteReq(const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
                          const char* pIdStr) {
   int32_t          totalRows = pDataBlock->info.rows;
@@ -90,6 +97,22 @@ end:
   return ret;
 }
 
+int32_t tqGetTableInfo(SSHashObj* tblInfo ,uint64_t groupId, STblInfo** pTbl) {
+  void* pVal = tSimpleHashGet(tblInfo, &groupId, sizeof(uint64_t));
+  if (pVal) {
+    *pTbl = *(STblInfo**)pVal;
+    return TSDB_CODE_SUCCESS;
+  }
+  return TSDB_CODE_FAILED;
+}
+
+int32_t tqPutTableInfo(SSHashObj* tblInfo ,uint64_t groupId, STblInfo* pTbl) {
+  if (tSimpleHashGetSize(tblInfo) > MAX_CATCH_NUM) {
+    return TSDB_CODE_SUCCESS;
+  }
+  return tSimpleHashPut(tblInfo, &groupId, sizeof(uint64_t), &pTbl, POINTER_BYTES);
+}
+
 int32_t tqPutReqToQueue(SVnode* pVnode, SVCreateTbBatchReq* pReqs) {
   void*   buf = NULL;
   int32_t tlen = 0;
@@ -260,100 +283,112 @@ void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* d
       tbData.suid = suid;
       tbData.uid = 0;  // uid is assigned by vnode
       tbData.sver = pTSchema->version;
+      STblInfo* pTblMeta = NULL;
 
-      char* ctbName = NULL;
-      tqDebug("vgId:%d, stream write into %s, table auto created", TD_VID(pVnode), pDataBlock->info.parTbName);
-      if (pDataBlock->info.parTbName[0]) {
-        ctbName = taosStrdup(pDataBlock->info.parTbName);
-      } else {
-        ctbName = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
+      int32_t res = tqGetTableInfo(pTask->tbSink.pTblInfo, pDataBlock->info.id.groupId, &pTblMeta);
+      if (res != TSDB_CODE_SUCCESS) {
+        pTblMeta = taosMemoryCalloc(1, sizeof(STblInfo));
       }
 
-      SMetaReader mr = {0};
-      metaReaderInit(&mr, pVnode->pMeta, 0);
-      if (metaGetTableEntryByName(&mr, ctbName) < 0) {
-        metaReaderClear(&mr);
-        tqDebug("vgId:%d, stream write into %s, table auto created", TD_VID(pVnode), ctbName);
+      char* ctbName = pDataBlock->info.parTbName;
+      if (!ctbName[0]) {
+        if (res == TSDB_CODE_SUCCESS) {
+          memcpy(ctbName, pTblMeta->tbName, strlen(pTblMeta->tbName));
+        } else {
+          char* tmp = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
+          memcpy(ctbName, tmp, strlen(tmp));
+          memcpy(pTblMeta->tbName, tmp, strlen(tmp));
+          taosMemoryFree(tmp);
+          tqDebug("vgId:%d, gropuid:%" PRIu64 " datablock tabel name is null", TD_VID(pVnode),
+                  pDataBlock->info.id.groupId);
+        }
+      }
 
-        SVCreateTbReq* pCreateTbReq = NULL;
+      if (res == TSDB_CODE_SUCCESS) {
+        tbData.uid = pTblMeta->uid;
+      } else {
+        SMetaReader mr = {0};
+        metaReaderInit(&mr, pVnode->pMeta, 0);
+        if (metaGetTableEntryByName(&mr, ctbName) < 0) {
+          metaReaderClear(&mr);
+          taosMemoryFree(pTblMeta);
+          tqDebug("vgId:%d, stream write into %s, table auto created", TD_VID(pVnode), ctbName);
 
-        if (!(pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateStbReq)))) {
-          taosMemoryFree(ctbName);
-          goto _end;
-        };
+          SVCreateTbReq* pCreateTbReq = NULL;
 
-        // set const
-        pCreateTbReq->flags = 0;
-        pCreateTbReq->type = TSDB_CHILD_TABLE;
-        pCreateTbReq->ctb.suid = suid;
+          if (!(pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateStbReq)))) {
+            goto _end;
+          };
 
-        // set super table name
-        SName name = {0};
-        tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
-        pCreateTbReq->ctb.stbName = taosStrdup((char*)tNameGetTableName(&name));  // taosStrdup(stbFullName);
+          // set const
+          pCreateTbReq->flags = 0;
+          pCreateTbReq->type = TSDB_CHILD_TABLE;
+          pCreateTbReq->ctb.suid = suid;
 
-        // set tag content
-        tagArray = taosArrayInit(1, sizeof(STagVal));
-        if (!tagArray) {
-          taosMemoryFree(ctbName);
-          tdDestroySVCreateTbReq(pCreateTbReq);
-          goto _end;
-        }
-        STagVal tagVal = {
-            .cid = pTSchema->numOfCols + 1,
-            .type = TSDB_DATA_TYPE_UBIGINT,
-            .i64 = (int64_t)pDataBlock->info.id.groupId,
-        };
-        taosArrayPush(tagArray, &tagVal);
-        pCreateTbReq->ctb.tagNum = taosArrayGetSize(tagArray);
+          // set super table name
+          SName name = {0};
+          tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
+          pCreateTbReq->ctb.stbName = taosStrdup((char*)tNameGetTableName(&name));  // taosStrdup(stbFullName);
 
-        STag* pTag = NULL;
-        tTagNew(tagArray, 1, false, &pTag);
-        tagArray = taosArrayDestroy(tagArray);
-        if (pTag == NULL) {
-          taosMemoryFree(ctbName);
-          tdDestroySVCreateTbReq(pCreateTbReq);
-          terrno = TSDB_CODE_OUT_OF_MEMORY;
-          taosMemoryFree(ctbName);
-          tdDestroySVCreateTbReq(pCreateTbReq);
-          goto _end;
-        }
-        pCreateTbReq->ctb.pTag = (uint8_t*)pTag;
+          // set tag content
+          tagArray = taosArrayInit(1, sizeof(STagVal));
+          if (!tagArray) {
+            tdDestroySVCreateTbReq(pCreateTbReq);
+            goto _end;
+          }
+          STagVal tagVal = {
+              .cid = pTSchema->numOfCols + 1,
+              .type = TSDB_DATA_TYPE_UBIGINT,
+              .i64 = (int64_t)pDataBlock->info.id.groupId,
+          };
+          taosArrayPush(tagArray, &tagVal);
+          pCreateTbReq->ctb.tagNum = taosArrayGetSize(tagArray);
 
-        // set tag name
-        SArray* tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
-        char    tagNameStr[TSDB_COL_NAME_LEN] = {0};
-        strcpy(tagNameStr, "group_id");
-        taosArrayPush(tagName, tagNameStr);
-        pCreateTbReq->ctb.tagName = tagName;
+          STag* pTag = NULL;
+          tTagNew(tagArray, 1, false, &pTag);
+          tagArray = taosArrayDestroy(tagArray);
+          if (pTag == NULL) {
+            tdDestroySVCreateTbReq(pCreateTbReq);
+            terrno = TSDB_CODE_OUT_OF_MEMORY;
+            goto _end;
+          }
+          pCreateTbReq->ctb.pTag = (uint8_t*)pTag;
 
-        // set table name
-        pCreateTbReq->name = ctbName;
-        ctbName = NULL;
+          // set tag name
+          SArray* tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
+          char    tagNameStr[TSDB_COL_NAME_LEN] = {0};
+          strcpy(tagNameStr, "group_id");
+          taosArrayPush(tagName, tagNameStr);
+          pCreateTbReq->ctb.tagName = tagName;
 
-        tbData.pCreateTbReq = pCreateTbReq;
-        tbData.flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
-      } else {
-        if (mr.me.type != TSDB_CHILD_TABLE) {
-          tqError("vgId:%d, failed to write into %s, since table type incorrect, type %d", TD_VID(pVnode), ctbName,
-                  mr.me.type);
-          metaReaderClear(&mr);
-          taosMemoryFree(ctbName);
-          continue;
-        }
+          // set table name
+          pCreateTbReq->name = taosStrdup(ctbName);
+
+          tbData.pCreateTbReq = pCreateTbReq;
+          tbData.flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
+        } else {
+          if (mr.me.type != TSDB_CHILD_TABLE) {
+            tqError("vgId:%d, failed to write into %s, since table type incorrect, type %d", TD_VID(pVnode), ctbName,
+                    mr.me.type);
+            metaReaderClear(&mr);
+            taosMemoryFree(pTblMeta);
+            continue;
+          }
+
+          if (mr.me.ctbEntry.suid != suid) {
+            tqError("vgId:%d, failed to write into %s, since suid mismatch, expect suid: %" PRId64
+                    ", actual suid %" PRId64 "",
+                    TD_VID(pVnode), ctbName, suid, mr.me.ctbEntry.suid);
+            metaReaderClear(&mr);
+            taosMemoryFree(pTblMeta);
+            continue;
+          }
 
-        if (mr.me.ctbEntry.suid != suid) {
-          tqError("vgId:%d, failed to write into %s, since suid mismatch, expect suid: %" PRId64
-                  ", actual suid %" PRId64 "",
-                  TD_VID(pVnode), ctbName, suid, mr.me.ctbEntry.suid);
+          tbData.uid = mr.me.uid;
+          pTblMeta->uid = mr.me.uid;
+          tqPutTableInfo(pTask->tbSink.pTblInfo, pDataBlock->info.id.groupId, pTblMeta);
           metaReaderClear(&mr);
-          taosMemoryFree(ctbName);
-          continue;
         }
-
-        tbData.uid = mr.me.uid;
-        metaReaderClear(&mr);
-        taosMemoryFreeClear(ctbName);
       }
 
       // rows

source/libs/executor/src/scanoperator.c

@@ -1583,7 +1583,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
   // currently only the tbname pseudo column
   if (pInfo->numOfPseudoExpr > 0) {
     int32_t code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes,
-                                          pInfo->pRes->info.rows, GET_TASKID(pTaskInfo), NULL);
+                                          pInfo->pRes->info.rows, GET_TASKID(pTaskInfo), &pTableScanInfo->base.metaCache);
     // ignore the table not exists error, since this table may have been dropped during the scan procedure.
     if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) {
       blockDataFreeRes((SSDataBlock*)pBlock);

source/libs/stream/src/streamTask.c

@@ -195,6 +195,7 @@ void tFreeStreamTask(SStreamTask* pTask) {
   if (pTask->outputType == TASK_OUTPUT__TABLE) {
     tDeleteSchemaWrapper(pTask->tbSink.pSchemaWrapper);
     taosMemoryFree(pTask->tbSink.pTSchema);
+    tSimpleHashCleanup(pTask->tbSink.pTblInfo);
   }
 
   if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {