Merge branch '3.0' into feature/TD-11463-3.0

source/libs/executor/inc/executorimpl.h

@@ -220,14 +220,9 @@ typedef struct SExecTaskInfo {
 } SExecTaskInfo;
 
 typedef struct STaskRuntimeEnv {
-
-  jmp_buf         env;
   STaskAttr*      pQueryAttr;
   uint32_t        status;  // query status
-  void*           qinfo;
   uint8_t         scanFlag;  // denotes reversed scan of data or not
-  void*           pTsdbReadHandle;
-  bool            enableGroupData;
   SDiskbasedBuf*  pResultBuf;           // query result buffer based on blocked-wised disk file
   SHashObj*       pResultRowHashTable;  // quick locate the window object for each result
   SHashObj*       pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
@@ -235,12 +230,10 @@ typedef struct STaskRuntimeEnv {
   char*           keyBuf;               // window key buffer
   // The window result objects pool, all the resultRow Objects are allocated and managed by this object.
   char**    prevRow;
-  SArray*   prevResult;  // intermediate result, SArray<SInterResult>
   STSBuf*   pTsBuf;      // timestamp filter list
   STSCursor cur;
 
   char*                          tagVal;  // tag value of current data block
-  struct SScalarFunctionSupport* scalarSup;
   STableGroupInfo tableqinfoGroupInfo;  // this is a group array list, including SArray<STableQueryInfo*> structure
   struct SOperatorInfo* proot;
   SGroupResInfo         groupResInfo;
@@ -266,7 +259,6 @@ typedef struct SOperatorInfo {
   char*                   name;          // name, used to show the query execution plan
   void*                   info;          // extension attribution
   SExprInfo*              pExpr;
-  STaskRuntimeEnv*        pRuntimeEnv;  // todo remove it
   SExecTaskInfo*          pTaskInfo;
   SOperatorCostInfo       cost;
   SResultInfo             resultInfo;
@@ -291,7 +283,7 @@ typedef struct {
 
 typedef enum {
   EX_SOURCE_DATA_NOT_READY = 0x1,
-  EX_SOURCE_DATA_READY = 0x2,
+  EX_SOURCE_DATA_READY     = 0x2,
   EX_SOURCE_DATA_EXHAUSTED = 0x3,
 } EX_SOURCE_STATUS;
 
@@ -682,7 +674,7 @@ SOperatorInfo* createTagScanOperatorInfo(SReaderHandle* pReaderHandle, SExprInfo
 void projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
                            int32_t numOfOutput, SArray* pPseudoList);
 
-void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order);
+void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, bool createDummyCol);
 
 void finalizeQueryResult(SqlFunctionCtx* pCtx, int32_t numOfOutput);
 void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);

source/libs/executor/src/groupoperator.c

@@ -284,7 +284,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgrou
     }
 
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, true);
 
     // there is an scalar expression that needs to be calculated right before apply the group aggregation.
     if (pInfo->pScalarExprInfo != NULL) {

source/libs/function/src/builtins.c

@@ -237,33 +237,45 @@ static int32_t translateLength(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, int32_t minParaNum, int32_t maxParaNum, int32_t primaryParaNo) {
+static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t len, int32_t minParaNum, int32_t maxParaNum, bool hasSep) {
   int32_t paraNum = LIST_LENGTH(pFunc->pParameterList);
   if (paraNum < minParaNum || paraNum > maxParaNum) {
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  uint8_t resultType = TSDB_DATA_TYPE_NCHAR;
+  uint8_t resultType = TSDB_DATA_TYPE_BINARY;
   int32_t resultBytes = 0;
   int32_t sepBytes = 0;
-  for (int32_t i = 0; i < LIST_LENGTH(pFunc->pParameterList); ++i) {
+
+  /* For concat/concat_ws function, if params have NCHAR type, promote the final result to NCHAR */
+  for (int32_t i = 0; i < paraNum; ++i) {
     SNode* pPara = nodesListGetNode(pFunc->pParameterList, i);
     uint8_t paraType = ((SExprNode*)pPara)->resType.type;
-    int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
     if (!IS_VAR_DATA_TYPE(paraType)) {
       return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
     }
-    if (i < primaryParaNo) {
-      sepBytes = paraBytes;
-      continue;
+    if (TSDB_DATA_TYPE_NCHAR == paraType) {
+      resultType = paraType;
+    }
+  }
+
+  for (int32_t i = 0; i < paraNum; ++i) {
+    SNode* pPara = nodesListGetNode(pFunc->pParameterList, i);
+    uint8_t paraType = ((SExprNode*)pPara)->resType.type;
+    int32_t paraBytes = ((SExprNode*)pPara)->resType.bytes;
+    int32_t factor = 1;
+    if (TSDB_DATA_TYPE_NCHAR == resultType && TSDB_DATA_TYPE_VARCHAR == paraType) {
+      factor *= TSDB_NCHAR_SIZE;
     }
-    if (TSDB_DATA_TYPE_BINARY == paraType) {
-      resultType = TSDB_DATA_TYPE_BINARY;
+    resultBytes += paraBytes * factor;
+
+    if (i == 0) {
+      sepBytes = paraBytes * factor;
     }
-    resultBytes += paraBytes;
   }
-  if (sepBytes > 0) {
-    resultBytes += sepBytes * (paraNum - 2);
+
+  if (hasSep) {
+    resultBytes += sepBytes * (paraNum - 3);
   }
 
   pFunc->node.resType = (SDataType) { .bytes = resultBytes, .type = resultType };
@@ -271,11 +283,11 @@ static int32_t translateConcatImpl(SFunctionNode* pFunc, char* pErrBuf, int32_t
 }
 
 static int32_t translateConcat(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
-  return translateConcatImpl(pFunc, pErrBuf, len, 2, 8, 0);
+  return translateConcatImpl(pFunc, pErrBuf, len, 2, 8, false);
 }
 
 static int32_t translateConcatWs(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
-  return translateConcatImpl(pFunc, pErrBuf, len, 3, 9, 1);
+  return translateConcatImpl(pFunc, pErrBuf, len, 3, 9, true);
 }
 
 static int32_t translateSubstr(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {

source/libs/nodes/src/nodesUtilFuncs.c

@@ -1226,7 +1226,7 @@ void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal) {
     case TSDB_DATA_TYPE_NCHAR:
     case TSDB_DATA_TYPE_VARCHAR:
     case TSDB_DATA_TYPE_VARBINARY:
-      pVal->pz = pNode->datum.p + VARSTR_HEADER_SIZE;
+      pVal->pz = pNode->datum.p;
       break;
     case TSDB_DATA_TYPE_JSON:
     case TSDB_DATA_TYPE_DECIMAL:

source/libs/scalar/src/sclfunc.c

@@ -313,8 +313,8 @@ static int32_t doLengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t concatCopyHelper(const char *input, char *output, bool hasNcharCol, int32_t type, int16_t *dataLen) {
-  if (hasNcharCol && type == TSDB_DATA_TYPE_VARCHAR) {
+static int32_t concatCopyHelper(const char *input, char *output, bool hasNchar, int32_t type, int16_t *dataLen) {
+  if (hasNchar && type == TSDB_DATA_TYPE_VARCHAR) {
     TdUcs4 *newBuf = taosMemoryCalloc((varDataLen(input) + 1) * TSDB_NCHAR_SIZE, 1);
     bool ret = taosMbsToUcs4(varDataVal(input), varDataLen(input), newBuf, (varDataLen(input) + 1) * TSDB_NCHAR_SIZE, NULL);
     if (!ret) {
@@ -345,10 +345,6 @@ static int32_t getNumOfNullEntries(SColumnInfoData *pColumnInfoData, int32_t num
 }
 
 int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
-  if (inputNum < 2 || inputNum > 8) { // concat accpet 2-8 input strings
-    return TSDB_CODE_FAILED;
-  }
-
   SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *));
   SColumnInfoData *pOutputData = pOutput->columnData;
   char **input = taosMemoryCalloc(inputNum, POINTER_BYTES);
@@ -356,15 +352,8 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
 
   int32_t inputLen = 0;
   int32_t numOfRows = 0;
-  bool hasNcharCol = false;
+  bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false;
   for (int32_t i = 0; i < inputNum; ++i) {
-    int32_t type = GET_PARAM_TYPE(&pInput[i]);
-    if (!IS_VAR_DATA_TYPE(type)) {
-      return TSDB_CODE_FAILED;
-    }
-    if (type == TSDB_DATA_TYPE_NCHAR) {
-      hasNcharCol = true;
-    }
     if (pInput[i].numOfRows > numOfRows) {
       numOfRows = pInput[i].numOfRows;
     }
@@ -373,7 +362,7 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
     pInputData[i] = pInput[i].columnData;
     input[i] = pInputData[i]->pData + pInputData[i]->varmeta.offset[0];
     int32_t factor = 1;
-    if (hasNcharCol && (GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_VARCHAR)) {
+    if (hasNchar && (GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_VARCHAR)) {
       factor = TSDB_NCHAR_SIZE;
     }
 
@@ -405,7 +394,7 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
 
     int16_t dataLen = 0;
     for (int32_t i = 0; i < inputNum; ++i) {
-      int32_t ret = concatCopyHelper(input[i], output, hasNcharCol, GET_PARAM_TYPE(&pInput[i]), &dataLen);
+      int32_t ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen);
       if (ret != TSDB_CODE_SUCCESS) {
         return ret;
       }
@@ -428,10 +417,6 @@ int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOu
 
 
 int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
-  if (inputNum < 3 || inputNum > 9) { // concat accpet 3-9 input strings including the separator
-    return TSDB_CODE_FAILED;
-  }
-
   SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *));
   SColumnInfoData *pOutputData = pOutput->columnData;
   char **input = taosMemoryCalloc(inputNum, POINTER_BYTES);
@@ -439,15 +424,8 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
 
   int32_t inputLen = 0;
   int32_t numOfRows = 0;
-  bool hasNcharCol = false;
+  bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false;
   for (int32_t i = 1; i < inputNum; ++i) {
-    int32_t type = GET_PARAM_TYPE(&pInput[i]);
-    if (!IS_VAR_DATA_TYPE(GET_PARAM_TYPE(&pInput[i]))) {
-      return TSDB_CODE_FAILED;
-    }
-    if (type == TSDB_DATA_TYPE_NCHAR) {
-      hasNcharCol = true;
-    }
     if (pInput[i].numOfRows > numOfRows) {
       numOfRows = pInput[i].numOfRows;
     }
@@ -456,7 +434,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
     pInputData[i] = pInput[i].columnData;
     input[i] = pInputData[i]->pData + pInputData[i]->varmeta.offset[0];
     int32_t factor = 1;
-    if (hasNcharCol && (GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_VARCHAR)) {
+    if (hasNchar && (GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_VARCHAR)) {
       factor = TSDB_NCHAR_SIZE;
     }
 
@@ -487,7 +465,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
         continue;
       }
 
-      int32_t ret = concatCopyHelper(input[i], output, hasNcharCol, GET_PARAM_TYPE(&pInput[i]), &dataLen);
+      int32_t ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen);
       if (ret != TSDB_CODE_SUCCESS) {
         return ret;
       }
@@ -499,7 +477,7 @@ int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
       if (i < inputNum - 1) {
         //insert the separator
         char *sep = pInputData[0]->pData;
-        int32_t ret = concatCopyHelper(sep, output, hasNcharCol, GET_PARAM_TYPE(&pInput[0]), &dataLen);
+        int32_t ret = concatCopyHelper(sep, output, hasNchar, GET_PARAM_TYPE(&pInput[0]), &dataLen);
         if (ret != TSDB_CODE_SUCCESS) {
           return ret;
         }