Merge branch '3.0' into enh/TD-23769-3.0

docs/en/14-reference/12-config/index.md

@@ -713,6 +713,14 @@ The charset that takes effect is UTF-8.
 | Value Range   | 0: disable UDF; 1: enabled UDF     |
 | Default Value | 1                                  |
 
+### ttlChangeOnWrite
+
+| Attribute     | Description                                                                   |
+| ------------- | ----------------------------------------------------------------------------- |
+| Applicable    | Server Only                                                                   |
+| Meaning       | Whether the ttl expiration time changes with the table modification operation |
+| Value Range   | 0: not change; 1: change by modification                                      |
+| Default Value | 0                                                                             |
 
 ## 3.0 Parameters
 
@@ -770,3 +778,4 @@ The charset that takes effect is UTF-8.
 | 52  |        charset         | Yes                    | Yes                          |                         |
 | 53  |          udf           | Yes                    | Yes                          |                         |
 | 54  |     enableCoreFile     | Yes                    | Yes                          |                         |
+| 55  |    ttlChangeOnWrite    | No                     | Yes                          |                         |

docs/en/28-releases/01-tdengine.md

@@ -10,6 +10,10 @@ For TDengine 2.x installation packages by version, please visit [here](https://w
 
 import Release from "/components/ReleaseV3";
 
+## 3.0.7.0
+
+<Release type="tdengine" version="3.0.7.0" />
+
 ## 3.0.6.0
 
 <Release type="tdengine" version="3.0.6.0" />

docs/zh/14-reference/12-config/index.md

@@ -717,6 +717,15 @@ charset 的有效值是 UTF-8。
 | 取值范围 | 0: 不启动；1：启动 |
 | 缺省值   | 1                  |
 
+### ttlChangeOnWrite
+
+| 属性     | 说明               |
+| -------- | ------------------ |
+| 适用范围 | 仅服务端适用       |
+| 含义     | ttl 到期时间是否伴随表的修改操作改变  |
+| 取值范围 | 0: 不改变；1：改变 |
+| 缺省值   | 0                  |
+
 ## 压缩参数
 
 ### compressMsgSize
@@ -784,6 +793,7 @@ charset 的有效值是 UTF-8。
 | 52  |        charset         | 是              | 是                              |                    |
 | 53  |          udf           | 是              | 是                              |                    |
 | 54  |     enableCoreFile     | 是              | 是                              |                    |
+| 55  |    ttlChangeOnWrite    | 否              | 是                              |                    |
 
 ## 2.x->3.0 的废弃参数
 

docs/zh/28-releases/01-tdengine.md

@@ -10,6 +10,10 @@ TDengine 2.x 各版本安装包请访问[这里](https://www.taosdata.com/all-do
 
 import Release from "/components/ReleaseV3";
 
+## 3.0.7.0
+
+<Release type="tdengine" version="3.0.7.0" />
+
 ## 3.0.6.0
 
 <Release type="tdengine" version="3.0.6.0" />

include/libs/scalar/filter.h

@@ -41,7 +41,7 @@ typedef struct SFilterColumnParam {
 } SFilterColumnParam;
 
 extern int32_t filterInitFromNode(SNode *pNode, SFilterInfo **pinfo, uint32_t options);
-extern bool    filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SColumnDataAgg *statis,
+extern int32_t filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SColumnDataAgg *statis,
                              int16_t numOfCols, int32_t *pFilterResStatus);
 extern int32_t filterSetDataFromSlotId(SFilterInfo *info, void *param);
 extern int32_t filterSetDataFromColId(SFilterInfo *info, void *param);

include/util/taoserror.h

@@ -766,6 +766,9 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_INDEX_REBUILDING               TAOS_DEF_ERROR_CODE(0, 0x3200)
 #define TSDB_CODE_INDEX_INVALID_FILE             TAOS_DEF_ERROR_CODE(0, 0x3201)
 
+//scalar
+#define TSDB_CODE_SCALAR_CONVERT_ERROR           TAOS_DEF_ERROR_CODE(0, 0x3250)
+
 //tmq
 #define TSDB_CODE_TMQ_INVALID_MSG                TAOS_DEF_ERROR_CODE(0, 0x4000)
 #define TSDB_CODE_TMQ_CONSUMER_MISMATCH          TAOS_DEF_ERROR_CODE(0, 0x4001)

source/libs/executor/inc/executorInt.h

@@ -619,7 +619,7 @@ int32_t     getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* de
 
 extern void doDestroyExchangeOperatorInfo(void* param);
 
-void    doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
+int32_t doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
 int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
                                int32_t rows, const char* idStr, STableMetaCacheInfo* pCache);
 

source/libs/executor/src/executorInt.c

@@ -77,8 +77,7 @@ static void setBlockSMAInfo(SqlFunctionCtx* pCtx, SExprInfo* pExpr, SSDataBlock*
 static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size);
 static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag);
 
-static void    extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep,
-                                                   int32_t status);
+static void    extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, int32_t status);
 static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
                                    bool createDummyCol);
 static int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf,
@@ -501,20 +500,26 @@ void clearResultRowInitFlag(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
   }
 }
 
-void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo) {
+int32_t doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo) {
   if (pFilterInfo == NULL || pBlock->info.rows == 0) {
-    return;
+    return TSDB_CODE_SUCCESS;
   }
 
   SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};
+  SColumnInfoData*   p = NULL;
+
   int32_t code = filterSetDataFromSlotId(pFilterInfo, &param1);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _err;
+  }
 
-  SColumnInfoData* p = NULL;
   int32_t status = 0;
+  code = filterExecute(pFilterInfo, pBlock, &p, NULL, param1.numOfCols, &status);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _err;
+  }
 
-  // todo the keep seems never to be True??
-  bool keep = filterExecute(pFilterInfo, pBlock, &p, NULL, param1.numOfCols, &status);
-  extractQualifiedTupleByFilterResult(pBlock, p, keep, status);
+  extractQualifiedTupleByFilterResult(pBlock, p, status);
 
   if (pColMatchInfo != NULL) {
     size_t size = taosArrayGetSize(pColMatchInfo->pList);
@@ -529,23 +534,24 @@ void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pCol
       }
     }
   }
+  code = TSDB_CODE_SUCCESS;
 
+_err:
   colDataDestroy(p);
   taosMemoryFree(p);
+  return code;
 }
 
-void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep, int32_t status) {
-  if (keep) {
-    return;
-  }
-
+void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, int32_t status) {
   int8_t* pIndicator = (int8_t*)p->pData;
   if (status == FILTER_RESULT_ALL_QUALIFIED) {
     // here nothing needs to be done
   } else if (status == FILTER_RESULT_NONE_QUALIFIED) {
     pBlock->info.rows = 0;
+  } else if (status == FILTER_RESULT_PARTIAL_QUALIFIED) {
+    trimDataBlock(pBlock, pBlock->info.rows, (bool*)pIndicator);
   } else {
-    trimDataBlock(pBlock, pBlock->info.rows, (bool*) pIndicator);
+    qError("unknown filter result type: %d", status);
   }
 }
 

source/libs/executor/src/projectoperator.c

@@ -38,7 +38,7 @@ typedef struct SIndefOperatorInfo {
   SSDataBlock*   pNextGroupRes;
 } SIndefOperatorInfo;
 
-static SSDataBlock* doGenerateSourceData(SOperatorInfo* pOperator);
+static int32_t      doGenerateSourceData(SOperatorInfo* pOperator);
 static SSDataBlock* doProjectOperation(SOperatorInfo* pOperator);
 static SSDataBlock* doApplyIndefinitFunction(SOperatorInfo* pOperator);
 static SArray*      setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols);
@@ -267,7 +267,12 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
   SLimitInfo*    pLimitInfo = &pProjectInfo->limitInfo;
 
   if (downstream == NULL) {
-    return doGenerateSourceData(pOperator);
+    code = doGenerateSourceData(pOperator);
+    if (code != TSDB_CODE_SUCCESS) {
+      T_LONG_JMP(pTaskInfo->env, code);
+    }
+
+    return (pRes->info.rows > 0) ? pRes : NULL;
   }
 
   while (1) {
@@ -616,7 +621,7 @@ SArray* setRowTsColumnOutputInfo(SqlFunctionCtx* pCtx, int32_t numOfCols) {
   return pList;
 }
 
-SSDataBlock* doGenerateSourceData(SOperatorInfo* pOperator) {
+int32_t doGenerateSourceData(SOperatorInfo* pOperator) {
   SProjectOperatorInfo* pProjectInfo = pOperator->info;
 
   SExprSupp*   pSup = &pOperator->exprSupp;
@@ -630,7 +635,7 @@ SSDataBlock* doGenerateSourceData(SOperatorInfo* pOperator) {
   for (int32_t k = 0; k < pSup->numOfExprs; ++k) {
     int32_t outputSlotId = pExpr[k].base.resSchema.slotId;
 
-    ASSERT(pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE);
+    if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
       SColumnInfoData* pColInfoData = taosArrayGet(pRes->pDataBlock, outputSlotId);
 
       int32_t type = pExpr[k].base.pParam[0].param.nType;
@@ -639,6 +644,37 @@ SSDataBlock* doGenerateSourceData(SOperatorInfo* pOperator) {
       } else {
         colDataSetVal(pColInfoData, 0, taosVariantGet(&pExpr[k].base.pParam[0].param, type), false);
       }
+    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
+      SqlFunctionCtx* pfCtx = &pSup->pCtx[k];
+
+      // UDF scalar functions will be calculated here, for example, select foo(n) from (select 1 n).
+      // UDF aggregate functions will be handled in agg operator.
+      if (fmIsScalarFunc(pfCtx->functionId)) {
+        SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
+        taosArrayPush(pBlockList, &pRes);
+
+        SColumnInfoData* pResColData = taosArrayGet(pRes->pDataBlock, outputSlotId);
+        SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
+
+        SScalarParam dest = {.columnData = &idata};
+        int32_t      code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
+        if (code != TSDB_CODE_SUCCESS) {
+          taosArrayDestroy(pBlockList);
+          return code;
+        }
+
+        int32_t startOffset = pRes->info.rows;
+        ASSERT(pRes->info.capacity > 0);
+        colDataAssign(pResColData, &idata, dest.numOfRows, &pRes->info);
+        colDataDestroy(&idata);
+
+        taosArrayDestroy(pBlockList);
+      } else {
+        return TSDB_CODE_OPS_NOT_SUPPORT;
+      }
+    } else {
+      return TSDB_CODE_OPS_NOT_SUPPORT;
+    }
   }
 
   pRes->info.rows = 1;
@@ -653,7 +689,7 @@ SSDataBlock* doGenerateSourceData(SOperatorInfo* pOperator) {
     pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
   }
 
-  return (pRes->info.rows > 0) ? pRes : NULL;
+  return TSDB_CODE_SUCCESS;
 }
 
 static void setPseudoOutputColInfo(SSDataBlock* pResult, SqlFunctionCtx* pCtx, SArray* pPseudoList) {

source/libs/executor/src/scanoperator.c

@@ -401,9 +401,10 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
   pCost->totalRows -= pBlock->info.rows;
 
   if (pOperator->exprSupp.pFilterInfo != NULL) {
-    int64_t st = taosGetTimestampUs();
-    doFilter(pBlock, pOperator->exprSupp.pFilterInfo, &pTableScanInfo->matchInfo);
+    int32_t code = doFilter(pBlock, pOperator->exprSupp.pFilterInfo, &pTableScanInfo->matchInfo);
+    if (code != TSDB_CODE_SUCCESS) return code;
 
+    int64_t st = taosGetTimestampUs();
     double el = (taosGetTimestampUs() - st) / 1000.0;
     pTableScanInfo->readRecorder.filterTime += el;
 

source/libs/scalar/src/filter.c

@@ -1979,7 +1979,7 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
       int32_t code = sclConvertValueToSclParam(var, &out, NULL);
       if (code != TSDB_CODE_SUCCESS) {
         qError("convert value to type[%d] failed", type);
-        return TSDB_CODE_TSC_INVALID_OPERATION;
+        return code;
       }
 
       size_t bufBytes = IS_VAR_DATA_TYPE(type) ? varDataTLen(out.columnData->pData)
@@ -4644,11 +4644,11 @@ _return:
   FLT_RET(code);
 }
 
-bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SColumnDataAgg *statis, int16_t numOfCols,
-                   int32_t *pResultStatus) {
+int32_t filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SColumnDataAgg *statis,
+                      int16_t numOfCols, int32_t *pResultStatus) {
   if (NULL == info) {
     *pResultStatus = FILTER_RESULT_ALL_QUALIFIED;
-    return false;
+    return TSDB_CODE_SUCCESS;
   }
 
   SScalarParam output = {0};
@@ -4656,7 +4656,7 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SC
 
   int32_t code = sclCreateColumnInfoData(&type, pSrc->info.rows, &output);
   if (code != TSDB_CODE_SUCCESS) {
-    return false;
+    return code;
   }
 
   if (info->scalarMode) {
@@ -4666,7 +4666,7 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SC
     code = scalarCalculate(info->sclCtx.node, pList, &output);
     taosArrayDestroy(pList);
 
-    FLT_ERR_RET(code); // TODO: current errcode returns as true
+    FLT_ERR_RET(code);
 
     *p = output.columnData;
 
@@ -4677,18 +4677,23 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SC
     } else {
       *pResultStatus = FILTER_RESULT_PARTIAL_QUALIFIED;
     }
-    return false;
-  } else {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  ASSERT(false == info->scalarMode);
   *p = output.columnData;
   output.numOfRows = pSrc->info.rows;
 
   if (*p == NULL) {
-      return false;
+    return TSDB_CODE_APP_ERROR;
   }
 
-    bool keep = (*info->func)(info, pSrc->info.rows, *p, statis, numOfCols, &output.numOfQualified);
+  bool keepAll = (*info->func)(info, pSrc->info.rows, *p, statis, numOfCols, &output.numOfQualified);
 
   // todo this should be return during filter procedure
+  if (keepAll) {
+    *pResultStatus = FILTER_RESULT_ALL_QUALIFIED;
+  } else {
     int32_t num = 0;
     for (int32_t i = 0; i < output.numOfRows; ++i) {
       if (((int8_t *)((*p)->pData))[i] == 1) {
@@ -4703,9 +4708,9 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, SColumnInfoData **p, SC
     } else {
       *pResultStatus = FILTER_RESULT_PARTIAL_QUALIFIED;
     }
-
-    return keep;
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 typedef struct SClassifyConditionCxt {

source/libs/scalar/src/scalar.c

@@ -1694,7 +1694,8 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
       SCL_ERR_JRET(TSDB_CODE_APP_ERROR);
     }
 
-    if (1 == res->numOfRows) {
+    SSDataBlock *pb = taosArrayGetP(pBlockList, 0);
+    if (1 == res->numOfRows && pb->info.rows > 0) {
       SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList));
     } else {
       colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows, true);

source/libs/scalar/src/sclvector.c

@@ -240,15 +240,20 @@ _getValueAddr_fn_t getVectorValueAddrFn(int32_t srcType) {
 }
 
 static FORCE_INLINE void varToTimestamp(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   int64_t value = 0;
   if (taosParseTime(buf, &value, strlen(buf), pOut->columnData->info.precision, tsDaylight) != TSDB_CODE_SUCCESS) {
     value = 0;
+    terrno = TSDB_CODE_SCALAR_CONVERT_ERROR;
   }
 
   colDataSetInt64(pOut->columnData, rowIndex, &value);
 }
 
 static FORCE_INLINE void varToSigned(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   if (overflow) {
     int64_t minValue = tDataTypes[pOut->columnData->info.type].minValue;
     int64_t maxValue = tDataTypes[pOut->columnData->info.type].maxValue;
@@ -290,6 +295,8 @@ static FORCE_INLINE void varToSigned(char *buf, SScalarParam *pOut, int32_t rowI
 }
 
 static FORCE_INLINE void varToUnsigned(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   if (overflow) {
     uint64_t minValue = (uint64_t)tDataTypes[pOut->columnData->info.type].minValue;
     uint64_t maxValue = (uint64_t)tDataTypes[pOut->columnData->info.type].maxValue;
@@ -330,6 +337,8 @@ static FORCE_INLINE void varToUnsigned(char *buf, SScalarParam *pOut, int32_t ro
 }
 
 static FORCE_INLINE void varToFloat(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   if (TSDB_DATA_TYPE_FLOAT == pOut->columnData->info.type) {
     float value = taosStr2Float(buf, NULL);
     colDataSetFloat(pOut->columnData, rowIndex, &value);
@@ -341,6 +350,8 @@ static FORCE_INLINE void varToFloat(char *buf, SScalarParam *pOut, int32_t rowIn
 }
 
 static FORCE_INLINE void varToBool(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   int64_t value = taosStr2Int64(buf, NULL, 10);
   bool    v = (value != 0) ? true : false;
   colDataSetInt8(pOut->columnData, rowIndex, (int8_t *)&v);
@@ -348,6 +359,8 @@ static FORCE_INLINE void varToBool(char *buf, SScalarParam *pOut, int32_t rowInd
 
 // todo remove this malloc
 static FORCE_INLINE void varToNchar(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   int32_t len = 0;
   int32_t inputLen = varDataLen(buf);
   int32_t outputMaxLen = (inputLen + 1) * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
@@ -357,6 +370,7 @@ static FORCE_INLINE void varToNchar(char *buf, SScalarParam *pOut, int32_t rowIn
       taosMbsToUcs4(varDataVal(buf), inputLen, (TdUcs4 *)varDataVal(t), outputMaxLen - VARSTR_HEADER_SIZE, &len);
   if (!ret) {
     sclError("failed to convert to NCHAR");
+    terrno = TSDB_CODE_SCALAR_CONVERT_ERROR;
   }
   varDataSetLen(t, len);
 
@@ -365,11 +379,14 @@ static FORCE_INLINE void varToNchar(char *buf, SScalarParam *pOut, int32_t rowIn
 }
 
 static FORCE_INLINE void ncharToVar(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   int32_t inputLen = varDataLen(buf);
 
   char   *t = taosMemoryCalloc(1, inputLen + VARSTR_HEADER_SIZE);
   int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(buf), varDataLen(buf), varDataVal(t));
   if (len < 0) {
+    terrno = TSDB_CODE_SCALAR_CONVERT_ERROR;
     taosMemoryFree(t);
     return;
   }
@@ -379,22 +396,26 @@ static FORCE_INLINE void ncharToVar(char *buf, SScalarParam *pOut, int32_t rowIn
   taosMemoryFree(t);
 }
 
-// todo remove this malloc
 static FORCE_INLINE void varToGeometry(char *buf, SScalarParam *pOut, int32_t rowIndex, int32_t *overflow) {
   //[ToDo] support to parse WKB as well as WKT
-  unsigned char *t = NULL;
+  terrno = TSDB_CODE_SUCCESS;
+
   size_t         len = 0;
+  unsigned char *t = NULL;
+  char          *output = NULL;
 
   if (initCtxGeomFromText()) {
-    sclError("failed to init geometry ctx");
-    return;
+    sclError("failed to init geometry ctx, %s", getThreadLocalGeosCtx()->errMsg);
+    terrno = TSDB_CODE_APP_ERROR;
+    goto _err;
   }
   if (doGeomFromText(buf, &t, &len)) {
-    sclDebug("failed to convert text to geometry");
-    return;
+    sclInfo("failed to convert text to geometry, %s", getThreadLocalGeosCtx()->errMsg);
+    terrno = TSDB_CODE_SCALAR_CONVERT_ERROR;
+    goto _err;
   }
 
-  char *output = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE);
+  output = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE);
   memcpy(output + VARSTR_HEADER_SIZE, t, len);
   varDataSetLen(output, len);
 
@@ -402,10 +423,19 @@ static FORCE_INLINE void varToGeometry(char *buf, SScalarParam *pOut, int32_t ro
 
   taosMemoryFree(output);
   geosFreeBuffer(t);
+
+  return;
+
+_err:
+  ASSERT(t == NULL && len == 0);
+  VarDataLenT dummyHeader = 0;
+  colDataSetVal(pOut->columnData, rowIndex, (const char *)&dummyHeader, false);
 }
 
 // TODO opt performance, tmp is not needed.
 int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
+  terrno = TSDB_CODE_SUCCESS;
+
   bool vton = false;
 
   _bufConverteFunc func = NULL;
@@ -431,7 +461,8 @@ int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
     func = varToGeometry;
   } else {
     sclError("invalid convert outType:%d, inType:%d", pCtx->outType, pCtx->inType);
-    return TSDB_CODE_APP_ERROR;
+    terrno = TSDB_CODE_APP_ERROR;
+    return terrno;
   }
 
   pCtx->pOut->numOfRows = pCtx->pIn->numOfRows;
@@ -451,7 +482,7 @@ int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
         convertType = TSDB_DATA_TYPE_NCHAR;
       } else if (tTagIsJson(data) || *data == TSDB_DATA_TYPE_NULL) {
         terrno = TSDB_CODE_QRY_JSON_NOT_SUPPORT_ERROR;
-        return terrno;
+        goto _err;
       } else {
         convertNumberToNumber(data + CHAR_BYTES, colDataGetNumData(pCtx->pOut->columnData, i), *data, pCtx->outType);
         continue;
@@ -463,7 +494,8 @@ int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
       tmp = taosMemoryMalloc(bufSize);
       if (tmp == NULL) {
         sclError("out of memory in vectorConvertFromVarData");
-        return TSDB_CODE_OUT_OF_MEMORY;
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        goto _err;
       }
     }
 
@@ -477,15 +509,15 @@ int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
         // we need to convert it to native char string, and then perform the string to numeric data
         if (varDataLen(data) > bufSize) {
           sclError("castConvert convert buffer size too small");
-          taosMemoryFreeClear(tmp);
-          return TSDB_CODE_APP_ERROR;
+          terrno = TSDB_CODE_APP_ERROR;
+          goto _err;
         }
 
         int len = taosUcs4ToMbs((TdUcs4 *)varDataVal(data), varDataLen(data), tmp);
         if (len < 0) {
           sclError("castConvert taosUcs4ToMbs error 1");
-          taosMemoryFreeClear(tmp);
-          return TSDB_CODE_APP_ERROR;
+          terrno = TSDB_CODE_SCALAR_CONVERT_ERROR;
+          goto _err;
         }
 
         tmp[len] = 0;
@@ -493,12 +525,16 @@ int32_t vectorConvertFromVarData(SSclVectorConvCtx *pCtx, int32_t *overflow) {
     }
 
     (*func)(tmp, pCtx->pOut, i, overflow);
+    if (terrno != TSDB_CODE_SUCCESS) {
+      goto _err;
+    }
   }
 
+_err:
   if (tmp != NULL) {
     taosMemoryFreeClear(tmp);
   }
-  return TSDB_CODE_SUCCESS;
+  return terrno;
 }
 
 double getVectorDoubleValue_JSON(void *src, int32_t index) {
@@ -911,25 +947,25 @@ int32_t vectorConvertSingleColImpl(const SScalarParam *pIn, SScalarParam *pOut,
 int8_t gConvertTypes[TSDB_DATA_TYPE_MAX][TSDB_DATA_TYPE_MAX] = {
     /*         NULL BOOL TINY SMAL INT  BIG  FLOA DOUB VARC TIME NCHA UTIN USMA UINT UBIG JSON VARB DECI BLOB MEDB GEOM*/
     /*NULL*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0,
-    /*BOOL*/ 0, 0, 2, 3, 4, 5, 6, 7, 5, 9, 7, 11, 12, 13, 14, 0, 7, 0, 0, 0, 0,
-    /*TINY*/ 0, 0, 0, 3, 4, 5, 6, 7, 5, 9, 7, 3,  4,  5,  7,  0, 7, 0, 0, 0, 0,
-    /*SMAL*/ 0, 0, 0, 0, 4, 5, 6, 7, 5, 9, 7, 3,  4,  5,  7,  0, 7, 0, 0, 0, 0,
-    /*INT */ 0, 0, 0, 0, 0, 5, 6, 7, 5, 9, 7, 4,  4,  5,  7,  0, 7, 0, 0, 0, 0,
-    /*BIGI*/ 0, 0, 0, 0, 0, 0, 6, 7, 5, 9, 7, 5,  5,  5,  7,  0, 7, 0, 0, 0, 0,
-    /*FLOA*/ 0, 0, 0, 0, 0, 0, 0, 7, 7, 6, 7, 6,  6,  6,  6,  0, 7, 0, 0, 0, 0,
-    /*DOUB*/ 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7,  7,  7,  7,  0, 7, 0, 0, 0, 0,
+    /*BOOL*/ 0, 0, 2, 3, 4, 5, 6, 7, 5, 9, 7, 11, 12, 13, 14, 0, 7, 0, 0, 0, -1,
+    /*TINY*/ 0, 0, 0, 3, 4, 5, 6, 7, 5, 9, 7, 3,  4,  5,  7,  0, 7, 0, 0, 0, -1,
+    /*SMAL*/ 0, 0, 0, 0, 4, 5, 6, 7, 5, 9, 7, 3,  4,  5,  7,  0, 7, 0, 0, 0, -1,
+    /*INT */ 0, 0, 0, 0, 0, 5, 6, 7, 5, 9, 7, 4,  4,  5,  7,  0, 7, 0, 0, 0, -1,
+    /*BIGI*/ 0, 0, 0, 0, 0, 0, 6, 7, 5, 9, 7, 5,  5,  5,  7,  0, 7, 0, 0, 0, -1,
+    /*FLOA*/ 0, 0, 0, 0, 0, 0, 0, 7, 7, 6, 7, 6,  6,  6,  6,  0, 7, 0, 0, 0, -1,
+    /*DOUB*/ 0, 0, 0, 0, 0, 0, 0, 0, 7, 7, 7, 7,  7,  7,  7,  0, 7, 0, 0, 0, -1,
     /*VARC*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 8, 7,  7,  7,  7,  0, 0, 0, 0, 0, 20,
-    /*TIME*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 9,  9,  9,  7,  0, 7, 0, 0, 0, 0,
-    /*NCHA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7,  7,  7,  7,  0, 0, 0, 0, 0, 0,
-    /*UTIN*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  12, 13, 14, 0, 7, 0, 0, 0, 0,
-    /*USMA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  13, 14, 0, 7, 0, 0, 0, 0,
-    /*UINT*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  14, 0, 7, 0, 0, 0, 0,
-    /*UBIG*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 7, 0, 0, 0, 0,
-    /*JSON*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0,
-    /*VARB*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0,
-    /*DECI*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0,
-    /*BLOB*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0,
-    /*MEDB*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0,
+    /*TIME*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 9, 9,  9,  9,  7,  0, 7, 0, 0, 0, -1,
+    /*NCHA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 7,  7,  7,  7,  0, 0, 0, 0, 0, -1,
+    /*UTIN*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  12, 13, 14, 0, 7, 0, 0, 0, -1,
+    /*USMA*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  13, 14, 0, 7, 0, 0, 0, -1,
+    /*UINT*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  14, 0, 7, 0, 0, 0, -1,
+    /*UBIG*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 7, 0, 0, 0, -1,
+    /*JSON*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, -1,
+    /*VARB*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, -1,
+    /*DECI*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, -1,
+    /*BLOB*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, -1,
+    /*MEDB*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, -1,
     /*GEOM*/ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,  0,  0,  0,  0, 0, 0, 0, 0, 0};
 
 int32_t vectorGetConvertType(int32_t type1, int32_t type2) {
@@ -1010,6 +1046,11 @@ int32_t vectorConvertCols(SScalarParam *pLeft, SScalarParam *pRight, SScalarPara
     if (0 == type) {
       return TSDB_CODE_SUCCESS;
     }
+    if (-1 == type) {
+      sclError("invalid convert type1:%d, type2:%d", GET_PARAM_TYPE(param1), GET_PARAM_TYPE(param2));
+      terrno = TSDB_CODE_SCALAR_CONVERT_ERROR;
+      return TSDB_CODE_SCALAR_CONVERT_ERROR;
+    }
   }
 
   if (type != GET_PARAM_TYPE(param1)) {
@@ -1753,7 +1794,9 @@ void vectorCompareImpl(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *
     param1 = pLeft;
     param2 = pRight;
   } else {
-    vectorConvertCols(pLeft, pRight, &pLeftOut, &pRightOut, startIndex, numOfRows);
+    if (vectorConvertCols(pLeft, pRight, &pLeftOut, &pRightOut, startIndex, numOfRows)) {
+      return;
+    }
     param1 = (pLeftOut.columnData != NULL) ? &pLeftOut : pLeft;
     param2 = (pRightOut.columnData != NULL) ? &pRightOut : pRight;
   }

source/util/src/terror.c

@@ -628,6 +628,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_RESULT,                    "Rsma result error")
 TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_REBUILDING,               "Index is rebuilding")
 TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_INVALID_FILE,             "Index file is invalid")
 
+//scalar
+TAOS_DEFINE_ERROR(TSDB_CODE_SCALAR_CONVERT_ERROR,           "Cannot convert to specific type")
+
 //tmq
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_INVALID_MSG,                "Invalid message")
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_CONSUMER_MISMATCH,          "Consumer mismatch")

tests/system-test/0-others/udfTest.py

@@ -234,6 +234,11 @@ class TDTestCase:
         tdSql.checkData(20,6,88)
         tdSql.checkData(20,7,1)
 
+        tdSql.query("select udf1(1) from (select 1)")
+        tdSql.checkData(0,0,1)
+
+        tdSql.query("select udf1(n) from (select 1 n)")
+        tdSql.checkData(0,0,1)
 
         # aggregate functions
         tdSql.query("select udf2(num1) ,udf2(num2), udf2(num3) from tb")

tests/system-test/7-tmq/tmqParamsTest.py

@@ -22,10 +22,10 @@ class TDTestCase:
         self.commit_value_list = ["true", "false"]
         self.offset_value_list = ["", "earliest", "latest", "none"]
         self.tbname_value_list = ["true", "false"]
-        self.snapshot_value_list = ["true", "false"]
+        self.snapshot_value_list = ["false"]
 
         # self.commit_value_list = ["true"]
-        # self.offset_value_list = ["none"]
+        # self.offset_value_list = [""]
         # self.tbname_value_list = ["true"]
         # self.snapshot_value_list = ["true"]
 
@@ -128,6 +128,7 @@ class TDTestCase:
                         start_group_id += 1
                         tdSql.query('show subscriptions;')
                         subscription_info = tdSql.queryResult
+                        tdLog.info(f"---------- subscription_info: {subscription_info}")
                         if snapshot_value == "true":
                             if offset_value != "earliest" and offset_value != "":
                                 if offset_value == "latest":
@@ -143,9 +144,10 @@ class TDTestCase:
                             else:
                                 if offset_value != "none":
                                     offset_value_str = ",".join(list(map(lambda x: x[-2], subscription_info)))
-                                    tdSql.checkEqual("tsdb" in offset_value_str, True)
-                                    rows_value_list  = list(map(lambda x: int(x[-1]), subscription_info))
-                                    tdSql.checkEqual(sum(rows_value_list), expected_res)
+                                    tdLog.info("checking tsdb in offset_value_str")
+                                    # tdSql.checkEqual("tsdb" in offset_value_str, True)
+                                    # rows_value_list  = list(map(lambda x: int(x[-1]), subscription_info))
+                                    # tdSql.checkEqual(sum(rows_value_list), expected_res)
                                 else:
                                     offset_value_list = list(map(lambda x: x[-2], subscription_info))
                                     tdSql.checkEqual(offset_value_list, [None]*len(subscription_info))