Merge branch '3.0' into fix/TD-31163-3.0

include/common/trow.h

@@ -310,7 +310,7 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
 int32_t tdGetTpRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int8_t colType, int32_t offset,
                            int16_t colIdx);
 int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int32_t offset, int16_t colIdx);
-void    tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColVal);
+int32_t tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColVal);
 
 typedef struct {
   STSchema *pSchema;

source/client/src/clientMsgHandler.c

@@ -576,8 +576,8 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {
   size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock) + PAYLOAD_PREFIX_LEN;
   *pRsp = taosMemoryCalloc(1, rspSize);
   if (NULL == *pRsp) {
-    blockDataDestroy(pBlock);
+    code = terrno;
-    return TSDB_CODE_OUT_OF_MEMORY;
+    goto  _exit;
   }
 
   (*pRsp)->useconds = 0;
@@ -589,6 +589,11 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {
   (*pRsp)->numOfCols = htonl(SHOW_VARIABLES_RESULT_COLS);
 
   int32_t len = blockEncode(pBlock, (*pRsp)->data + PAYLOAD_PREFIX_LEN, SHOW_VARIABLES_RESULT_COLS);
+  if(len < 0) {
+    uError("buildShowVariablesRsp error, len:%d", len);
+    code = terrno;
+    goto _exit;
+  }
   blockDataDestroy(pBlock);
 
   SET_PAYLOAD_LEN((*pRsp)->data, len, len);
@@ -600,10 +605,21 @@ static int32_t buildShowVariablesRsp(SArray* pVars, SRetrieveTableRsp** pRsp) {
   if (payloadLen != rspSize - sizeof(SRetrieveTableRsp)) {
     uError("buildShowVariablesRsp error, len:%d != rspSize - sizeof(SRetrieveTableRsp):%" PRIu64, len,
            (uint64_t)(rspSize - sizeof(SRetrieveTableRsp)));
-    return TSDB_CODE_TSC_INVALID_INPUT;
+    code = TSDB_CODE_TSC_INVALID_INPUT;
+    goto _exit;
   }
 
   return TSDB_CODE_SUCCESS;
+_exit:
+  if(*pRsp)  {
+    taosMemoryFree(*pRsp);
+    *pRsp = NULL;
+  }
+  if(pBlock) {
+    blockDataDestroy(pBlock);
+    pBlock = NULL;
+  }
+  return code;
 }
 
 int32_t processShowVariablesRsp(void* param, SDataBuf* pMsg, int32_t code) {
@@ -711,8 +727,8 @@ static int32_t buildRetriveTableRspForCompactDb(SCompactDbRsp* pCompactDb, SRetr
   size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock) + PAYLOAD_PREFIX_LEN;
   *pRsp = taosMemoryCalloc(1, rspSize);
   if (NULL == *pRsp) {
-    blockDataDestroy(pBlock);
+    code = terrno;
-    return TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
   }
 
   (*pRsp)->useconds = 0;
@@ -725,6 +741,11 @@ static int32_t buildRetriveTableRspForCompactDb(SCompactDbRsp* pCompactDb, SRetr
   (*pRsp)->numOfCols = htonl(COMPACT_DB_RESULT_COLS);
 
   int32_t len = blockEncode(pBlock, (*pRsp)->data + PAYLOAD_PREFIX_LEN, COMPACT_DB_RESULT_COLS);
+  if(len < 0) {
+    uError("buildRetriveTableRspForCompactDb error, len:%d", len);
+    code = terrno;
+    goto _exit;
+  }
   blockDataDestroy(pBlock);
 
   SET_PAYLOAD_LEN((*pRsp)->data, len, len);
@@ -736,10 +757,21 @@ static int32_t buildRetriveTableRspForCompactDb(SCompactDbRsp* pCompactDb, SRetr
   if (payloadLen != rspSize - sizeof(SRetrieveTableRsp)) {
     uError("buildRetriveTableRspForCompactDb error, len:%d != rspSize - sizeof(SRetrieveTableRsp):%" PRIu64, len,
            (uint64_t)(rspSize - sizeof(SRetrieveTableRsp)));
-    return TSDB_CODE_TSC_INVALID_INPUT;
+    code = TSDB_CODE_TSC_INVALID_INPUT;
+    goto _exit;
   }
 
   return TSDB_CODE_SUCCESS;
+_exit:
+  if(*pRsp)  {
+    taosMemoryFree(*pRsp);
+    *pRsp = NULL;
+  }
+  if(pBlock) {
+    blockDataDestroy(pBlock);
+    pBlock = NULL;
+  }
+  return code;
 }
 
 

source/common/src/tdatablock.c

@@ -2888,6 +2888,7 @@ int32_t buildCtbNameByGroupIdImpl(const char* stbFullName, uint64_t groupId, cha
   return code;
 }
 
+// return length of encoded data, return -1 if failed
 int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
   int32_t dataLen = 0;
 
@@ -2921,7 +2922,7 @@ int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
     if (pColInfoData == NULL) {
-      return terrno;
+      return -1;
     }
 
     *((int8_t*)data) = pColInfoData->info.type;
@@ -2940,7 +2941,7 @@ int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
   for (int32_t col = 0; col < numOfCols; ++col) {
     SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, col);
     if (pColRes == NULL) {
-      return terrno;
+      return -1;
     }
 
     // copy the null bitmap
@@ -2991,7 +2992,6 @@ int32_t blockEncode(const SSDataBlock* pBlock, char* data, int32_t numOfCols) {
 
   *actualLen = dataLen;
   *groupId = pBlock->info.id.groupId;
-  ASSERT(dataLen > 0);
   return dataLen;
 }
 

source/common/src/trow.c

@@ -232,7 +232,6 @@ void tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag) {
     if (!tdSTSRowIterNext(&iter, &sVal)) {
       break;
     }
-    ASSERT(sVal.valType == 0 || sVal.valType == 1 || sVal.valType == 2);
     tdSCellValPrint(&sVal, cols[iter.colIdx - 1].type);
   }
   printf("\n");
@@ -389,7 +388,7 @@ bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal) {
   } else if (TD_IS_KV_ROW(pIter->pRow)) {
     (void)tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal);
   } else {
-    ASSERT(0);
+    return false;
   }
   ++pIter->colIdx;
 
@@ -409,7 +408,9 @@ int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow, int8_t r
   void     *varBuf = NULL;
   bool      isAlloc = false;
 
-  ASSERT(nColVal > 1);
+  if(nColVal <= 1) {
+    TAOS_RETURN(TSDB_CODE_INVALID_PARA);
+  }
 
   for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; ++iColumn) {
     pTColumn = &pTSchema->columns[iColumn];
@@ -423,9 +424,10 @@ int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow, int8_t r
     }
 
     if (iColumn == 0) {
-      ASSERT(pColVal && pColVal->cid == pTColumn->colId);
+      if ((pColVal && pColVal->cid != pTColumn->colId) || (pTColumn->type != TSDB_DATA_TYPE_TIMESTAMP) ||
-      ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
+          (pTColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID)) {
-      ASSERT(pTColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
+        TAOS_RETURN(TSDB_CODE_INVALID_PARA);
+      }
     } else {
       if (IS_VAR_DATA_TYPE(pTColumn->type)) {
         if (pColVal && COL_VAL_IS_VALUE(pColVal)) {
@@ -598,7 +600,10 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
 
 int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int32_t offset, int16_t colIdx) {
 #ifdef TD_SUPPORT_BITMAP
-  ASSERT(colIdx < tdRowGetNCols(pRow) - 1);
+  if (!(colIdx < tdRowGetNCols(pRow) - 1)) {
+    output->valType = TD_VTYPE_NONE;
+    TAOS_RETURN(TSDB_CODE_INVALID_PARA);
+  }
   int32_t code = 0;
   if ((code = tdGetBitmapValType(pBitmap, colIdx, &output->valType, 0)) != TSDB_CODE_SUCCESS) {
     output->valType = TD_VTYPE_NONE;
@@ -699,7 +704,6 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
 int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val, bool isCopyVarData,
                               int8_t colType, int16_t colIdx, int32_t offset, col_id_t colId) {
   if (colIdx < 1) {
-    ASSERTS(0, "colIdx is %" PRIi64, colIdx);
     TAOS_RETURN(TSDB_CODE_INVALID_PARA);
   }
   --colIdx;
@@ -775,7 +779,9 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
   TD_ROW_SET_INFO(pBuilder->pBuf, 0);
   TD_ROW_SET_TYPE(pBuilder->pBuf, pBuilder->rowType);
 
-  ASSERT(pBuilder->nBitmaps > 0 && pBuilder->flen > 0);
+  if(!(pBuilder->nBitmaps > 0 && pBuilder->flen > 0)) {
+    TAOS_RETURN(TSDB_CODE_INVALID_PARA);
+  }
 
   uint32_t len = 0;
   switch (pBuilder->rowType) {
@@ -814,7 +820,9 @@ int32_t tdSRowGetBuf(SRowBuilder *pBuilder, void *pBuf) {
     TAOS_RETURN(TSDB_CODE_INVALID_PARA);
   }
 
-  ASSERT(pBuilder->nBitmaps > 0 && pBuilder->flen > 0);
+  if(!(pBuilder->nBitmaps > 0 && pBuilder->flen > 0)) {
+    TAOS_RETURN(TSDB_CODE_INVALID_PARA);
+  }
 
   uint32_t len = 0;
   switch (pBuilder->rowType) {
@@ -930,18 +938,20 @@ int32_t tdSetBitmapValType(void *pBitmap, int16_t colIdx, TDRowValT valType, int
   TAOS_RETURN(TSDB_CODE_SUCCESS);
 }
 
-void tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColVal) {
+int32_t tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColVal) {
   STColumn *pTColumn = &pTSchema->columns[iCol];
   SCellVal  cv = {0};
 
-  ASSERT((pTColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID) || (iCol > 0));
+  if (!((pTColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID) || (iCol > 0))) {
+    TAOS_RETURN(TSDB_CODE_INVALID_PARA);
+  }
 
   if (TD_IS_TP_ROW(pRow)) {
     (void)tdSTpRowGetVal(pRow, pTColumn->colId, pTColumn->type, pTSchema->flen, pTColumn->offset, iCol - 1, &cv);
   } else if (TD_IS_KV_ROW(pRow)) {
     (void)tdSKvRowGetVal(pRow, pTColumn->colId, iCol - 1, &cv);
   } else {
-    ASSERT(0);
+    TAOS_RETURN(TSDB_CODE_INVALID_PARA);
   }
 
   if (tdValTypeIsNone(cv.valType)) {
@@ -960,4 +970,5 @@ void tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColV
       (void)memcpy(&pColVal->value.val, cv.val, tDataTypes[pTColumn->type].bytes);
     }
   }
+  return 0;
 }

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -479,6 +479,12 @@ int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   }
 
   int32_t len = blockEncode(pBlock, pStart, numOfCols);
+  if(len < 0) {
+    dError("failed to retrieve data since %s", tstrerror(code));
+    blockDataDestroy(pBlock);
+    rpcFreeCont(pRsp);
+    return terrno;
+  }
 
   pRsp->numOfRows = htonl(pBlock->info.rows);
   pRsp->precision = TSDB_TIME_PRECISION_MILLI;  // millisecond time precision

source/dnode/mnode/impl/src/mndShow.c

@@ -330,11 +330,9 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
 
   SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
   if (pRsp == NULL) {
-    mndReleaseShowObj(pShow, false);
-    code = TSDB_CODE_OUT_OF_MEMORY;
     mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, tstrerror(code));
-    blockDataDestroy(pBlock);
+    code = terrno;
-    TAOS_RETURN(code);
+    goto _exit;
   }
 
   pRsp->handle = htobe64(pShow->id);
@@ -356,6 +354,11 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
     }
 
     int32_t len = blockEncode(pBlock, pStart, pShow->pMeta->numOfColumns);
+    if(len < 0){
+      mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, tstrerror(code));
+      code =  terrno;
+      return code;
+    }
   }
 
   pRsp->numOfRows = htonl(rowsRead);
@@ -374,6 +377,13 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
 
   blockDataDestroy(pBlock);
   return TSDB_CODE_SUCCESS;
+_exit:
+  mndReleaseShowObj(pShow, false);
+  blockDataDestroy(pBlock);
+  if(pRsp) {
+    rpcFreeCont(pRsp);
+  }
+  return code;
 }
 
 static bool mndCheckRetrieveFinished(SShowObj *pShow) {

source/dnode/mnode/impl/src/mndSync.c

@@ -189,13 +189,16 @@ int32_t mndProcessWriteMsg(SMnode *pMnode, SRpcMsg *pMsg, SFsmCbMeta *pMeta) {
     goto _OUT;
   }
 
+  (void)taosThreadMutexLock(&pMnode->pSdb->filelock);
   code = sdbWriteWithoutFree(pMnode->pSdb, pRaw);
   if (code != 0) {
     mError("trans:%d, failed to write to sdb since %s", transId, terrstr());
     // code = 0;
+    (void)taosThreadMutexUnlock(&pMnode->pSdb->filelock);
     pMeta->code = code;
     goto _OUT;
   }
+  (void)taosThreadMutexUnlock(&pMnode->pSdb->filelock);
 
   pTrans = mndAcquireTrans(pMnode, transId);
   if (pTrans == NULL) {

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

@@ -928,7 +928,7 @@ static int32_t tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid, SRSmaInfo **ppR
 
     tdRefRSmaInfo(pSma, pRSmaInfo);
     taosRUnLockLatch(SMA_ENV_LOCK(pEnv));
-    if (ASSERTS(pRSmaInfo->suid == suid, "suid:%" PRIi64 " != %" PRIi64, pRSmaInfo->suid, suid)) {
+    if (pRSmaInfo->suid != suid) {
       TAOS_RETURN(TSDB_CODE_APP_ERROR);
     }
     *ppRSmaInfo = pRSmaInfo;
@@ -1645,7 +1645,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
                     ((oldStat == 2) && atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat)) < TASK_TRIGGER_STAT_PAUSED)) {
                   int32_t oldVal = atomic_fetch_add_32(&pRSmaStat->nFetchAll, 1);
 
-                  if (ASSERTS(oldVal >= 0, "oldVal of nFetchAll: %d < 0", oldVal)) {
+                  if (oldVal < 0) {
                     code = TSDB_CODE_APP_ERROR;
                     taosHashCancelIterate(infoHash, pIter);
                     TSDB_CHECK_CODE(code, lino, _exit);

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

@@ -350,8 +350,7 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
     }
   }
 
-  if (ASSERTS(pTsmaStat->pTSma->indexUid == indexUid, "indexUid:%" PRIi64 " != %" PRIi64, pTsmaStat->pTSma->indexUid,
+  if (pTsmaStat->pTSma->indexUid != indexUid) {
-              indexUid)) {
     code = TSDB_CODE_APP_ERROR;
     TSDB_CHECK_CODE(code, lino, _exit);
   }

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

@@ -29,6 +29,10 @@ int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, void* pRsp, int32_t numOf
   pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);
 
   int32_t actualLen = blockEncode(pBlock, pRetrieve->data, numOfCols);
+  if(actualLen < 0){
+    taosMemoryFree(buf);
+    return terrno;
+  }
   actualLen += sizeof(SRetrieveTableRspForTmq);
   if (taosArrayPush(((SMqDataRspCommon*)pRsp)->blockDataLen, &actualLen) == NULL){
     taosMemoryFree(buf);

source/libs/command/src/command.c

@@ -50,6 +50,10 @@ static int32_t buildRetrieveTableRsp(SSDataBlock* pBlock, int32_t numOfCols, SRe
   (*pRsp)->numOfCols = htonl(numOfCols);
 
   int32_t len = blockEncode(pBlock, (*pRsp)->data + PAYLOAD_PREFIX_LEN, numOfCols);
+  if(len < 0) {
+    taosMemoryFree(*pRsp);
+    return terrno;
+  }
   SET_PAYLOAD_LEN((*pRsp)->data, len, len);
 
   int32_t payloadLen = len + PAYLOAD_PREFIX_LEN;

source/libs/command/src/explain.c

@@ -1978,6 +1978,10 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
   rsp->numOfRows = htobe64((int64_t)rowNum);
 
   int32_t len = blockEncode(pBlock, rsp->data + PAYLOAD_PREFIX_LEN, taosArrayGetSize(pBlock->pDataBlock));
+  if(len < 0) {
+    qError("qExplainGetRspFromCtx: blockEncode failed");
+    QRY_ERR_JRET(terrno);
+  }
 
   rsp->compLen = htonl(len);
   rsp->payloadLen = htonl(len);

source/libs/executor/src/dataDispatcher.c

@@ -106,6 +106,10 @@ static int32_t toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData*
       }
 
       int32_t dataLen = blockEncode(pInput->pData, pHandle->pCompressBuf, numOfCols);
+      if(dataLen < 0) {
+        qError("failed to encode data block, code: %d", dataLen);
+        return terrno;
+      }
       int32_t len =
           tsCompressString(pHandle->pCompressBuf, dataLen, 1, pEntry->data, pBuf->allocSize, ONE_STAGE_COMP, NULL, 0);
       if (len < dataLen) {
@@ -120,6 +124,10 @@ static int32_t toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData*
       }
     } else {
       pEntry->dataLen = blockEncode(pInput->pData, pEntry->data, numOfCols);
+      if(pEntry->dataLen < 0) {
+        qError("failed to encode data block, code: %d", pEntry->dataLen);
+        return terrno;
+      }
       pEntry->rawLen = pEntry->dataLen;
     }
   }

source/libs/executor/src/groupoperator.c

@@ -548,7 +548,7 @@ int32_t createGroupOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNo
   SGroupbyOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SGroupbyOperatorInfo));
   SOperatorInfo*        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
-    code = TSDB_CODE_OUT_OF_MEMORY;
+    code = terrno;
     goto _error;
   }
 

source/libs/scalar/src/sclfunc.c

@@ -1544,9 +1544,15 @@ int32_t replaceFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pO
       needFreeFrom = true;
     }
     if (GET_PARAM_TYPE(&pInput[2]) != GET_PARAM_TYPE(&pInput[0])) {
-      SCL_ERR_JRET(convBetweenNcharAndVarchar(varDataVal(colDataGetData(pInputData[2], colIdx3)), &toStr,
+      code = convBetweenNcharAndVarchar(varDataVal(colDataGetData(pInputData[2], colIdx3)), &toStr,
-                                             varDataLen(colDataGetData(pInputData[2], colIdx3)), &toLen,
+                                        varDataLen(colDataGetData(pInputData[2], colIdx3)), &toLen,
-                                             GET_PARAM_TYPE(&pInput[0])));
+                                        GET_PARAM_TYPE(&pInput[0]));
+      if (TSDB_CODE_SUCCESS != code) {
+        if (needFreeFrom) {
+          taosMemoryFree(fromStr);
+        }
+        goto _return;
+      }
       needFreeTo = true;
     }
 
@@ -1660,9 +1666,9 @@ int32_t substrIdxFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *
 
     SCL_ERR_JRET(colDataSetVal(pOutputData, k, output, false));
   }
-
-_return:
   pOutput->numOfRows = numOfRows;
+_return:
+  taosMemoryFree(outputBuf);
   return code;
 }
 

source/libs/stream/src/streamDispatch.c

@@ -158,6 +158,10 @@ static int32_t buildStreamRetrieveReq(SStreamTask* pTask, const SSDataBlock* pBl
   pRetrieve->version = htobe64(pBlock->info.version);
 
   int32_t actualLen = blockEncode(pBlock, pRetrieve->data + PAYLOAD_PREFIX_LEN, numOfCols);
+  if(actualLen < 0) {
+    taosMemoryFree(pRetrieve);
+    return terrno;
+  }
   SET_PAYLOAD_LEN(pRetrieve->data, actualLen, actualLen);
 
   int32_t payloadLen = actualLen + PAYLOAD_PREFIX_LEN;
@@ -1064,7 +1068,7 @@ int32_t streamAddBlockIntoDispatchMsg(const SSDataBlock* pBlock, SStreamDispatch
 
   void* buf = taosMemoryCalloc(1, dataStrLen);
   if (buf == NULL) {
-    return -1;
+    return terrno;
   }
 
   SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
@@ -1084,6 +1088,10 @@ int32_t streamAddBlockIntoDispatchMsg(const SSDataBlock* pBlock, SStreamDispatch
   pRetrieve->numOfCols = htonl(numOfCols);
 
   int32_t actualLen = blockEncode(pBlock, pRetrieve->data + PAYLOAD_PREFIX_LEN, numOfCols);
+  if(actualLen < 0) {
+    taosMemoryFree(buf);
+    return terrno;
+  }
   SET_PAYLOAD_LEN(pRetrieve->data, actualLen, actualLen);
 
   int32_t payloadLen = actualLen + PAYLOAD_PREFIX_LEN;

tests/system-test/2-query/concat.py

@@ -144,6 +144,9 @@ class TDTestCase:
             f"{dbname}.ct2",
             f"{dbname}.ct4",
         ]
+        tdSql.query("select concat(null,null)")  # TD-31571
+        tdSql.checkRows(1)
+        tdSql.checkData(0, 0, None)
         for tb in tbname:
             for i in range(2,8):
                 self.__concat_check(tb,i)

tests/system-test/2-query/concat_ws.py

@@ -156,7 +156,9 @@ class TDTestCase:
             f"{dbname}.ct2",
             f"{dbname}.ct4",
         ]
-
+        tdSql.query("select concat_ws(null,null,null);")  # TD-31572
+        tdSql.checkRows(1)
+        tdSql.checkData(0, 0, None)
         for tb in tbname:
             for errsql in self.__concat_ws_err_check(tb):
                 tdSql.error(sql=errsql)