chore: add debug info

include/common/tdatablock.h

@@ -244,7 +244,7 @@ const char* blockDecode(SSDataBlock* pBlock, const char* pData);
 char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 
 int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId,
-                                    tb_uid_t suid);
+                                    tb_uid_t suid, int64_t blkVer, const char* tag);
 
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
 int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf);

source/common/src/tdatablock.c

@@ -1872,13 +1872,32 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
   return dumpBuf;
 }
 
+static SHashObj* dupCheck = NULL;
+static int8_t    dupInit = 0;
+
 int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDataBlock, const STSchema* pTSchema,
-                                    int64_t uid, int32_t vgId, tb_uid_t suid) {
+                                    int64_t uid, int32_t vgId, tb_uid_t suid, int64_t blkVer, const char* tag) {
   SSubmitReq2* pReq = *ppReq;
   SArray*      pVals = NULL;
   int32_t      numOfBlks = 0;
   int32_t      sz = 1;
 
+  if (!dupCheck) {
+    if (0 == atomic_val_compare_exchange_8(&dupInit, 0, 1)) {
+      dupCheck = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
+      if (!dupCheck) ASSERT(0);
+    } else {
+      int32_t cnt = 0;
+      while (!dupCheck) {
+        ++cnt;
+        if (cnt > 1000) {
+          sched_yield();
+          cnt = 0;
+        }
+      }
+    }
+  }
+
   terrno = TSDB_CODE_SUCCESS;
 
   if (NULL == pReq) {
@@ -1892,6 +1911,8 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
     }
   }
 
+  char dupKey[40];
+
   for (int32_t i = 0; i < sz; ++i) {
     int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
     int32_t rows = pDataBlock->info.rows;
@@ -1936,6 +1957,18 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDat
               ASSERT(PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId);
               SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, (SValue){.val = *(TSKEY*)var});
               taosArrayPush(pVals, &cv);
+              snprintf(dupKey, 40, "%" PRIi64 ":%" PRIi64, *(TSKEY*)var, blkVer);
+              uInfo("%s:%d key:ver: %s, tags: %s", __func__, __LINE__, dupKey, tag);
+              int32_t dupKeyLen = strlen(dupKey);
+              assert(dupKeyLen < 40);
+              void* hashKey = NULL;
+              if ((hashKey = taosHashGet(dupCheck, &dupKey, dupKeyLen + 1))) {
+                ASSERT(0);
+              } else {
+                if(taosHashPut(dupCheck, &dupKey, dupKeyLen + 1, NULL, 0) != 0){
+                  ASSERT(0);
+                }
+              }
             } else if (colDataIsNull_s(pColInfoData, j)) {
               SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
               taosArrayPush(pVals, &cv);

source/dnode/vnode/src/sma/smaRollup.c

@@ -45,7 +45,7 @@ static void       tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
 static void       tdFreeRSmaSubmitItems(SArray *pItems);
 static int32_t    tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo);
 static int32_t    tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
-                                            int64_t suid);
+                                            int64_t suid, const char* tag);
 static void       tdRSmaFetchTrigger(void *param, void *tmrId);
 static void       tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
 static int32_t    tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables);
@@ -581,7 +581,7 @@ _end:
 }
 
 static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema,
-                                         int64_t suid) {
+                                         int64_t suid, const char* tag) {
   int32_t      code = 0;
   int32_t      lino = 0;
   SSDataBlock *output = NULL;
@@ -620,7 +620,8 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
       SSubmitReq2 *pReq = NULL;
 
       // TODO: the schema update should be handled later(TD-17965)
-      if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, output->info.id.groupId, SMA_VID(pSma), suid) < 0) {
+      if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, output->info.id.groupId, SMA_VID(pSma), suid,
+                                      output->info.version, tag) < 0) {
         code = terrno ? terrno : TSDB_CODE_RSMA_RESULT;
         TSDB_CHECK_CODE(code, lino, _exit);
       }
@@ -775,7 +776,7 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize,
   }
 
   SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, idx);
-  tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo->pTSchema, pInfo->suid);
+  tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo->pTSchema, pInfo->suid, __func__);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -1246,7 +1247,7 @@ static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) {
       if ((terrno = qSetSMAInput(taskInfo, &dataBlock, 1, STREAM_INPUT__DATA_BLOCK)) < 0) {
         goto _err;
       }
-      if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo->pTSchema, pInfo->suid) < 0) {
+      if (tdRSmaExecAndSubmitResult(pSma, taskInfo, pItem, pInfo->pTSchema, pInfo->suid, __func__) < 0) {
         goto _err;
       }