Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/tsdb_snapshot

include/common/ttime.h

@@ -72,6 +72,8 @@ static FORCE_INLINE int64_t taosGetTimestampToday(int32_t precision) {
 }
 
 int64_t taosTimeAdd(int64_t t, int64_t duration, char unit, int32_t precision);
+int64_t taosTimeSub(int64_t t, int64_t duration, char unit, int32_t precision);
+
 int64_t taosTimeTruncate(int64_t t, const SInterval* pInterval, int32_t precision);
 int32_t taosTimeCountInterval(int64_t skey, int64_t ekey, int64_t interval, char unit, int32_t precision);
 

include/libs/scalar/scalar.h

@@ -32,6 +32,7 @@ pNode will be freed in API;
 *pRes need to freed in caller
 */
 int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes);
+int32_t scalarCalculateConstantsFromDual(SNode *pNode, SNode **pRes);
 
 /*
 pDst need to freed in caller

include/util/taoserror.h

@@ -583,7 +583,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_INTERP_CLAUSE     TAOS_DEF_ERROR_CODE(0, 0x265D)
 #define TSDB_CODE_PAR_NO_VALID_FUNC_IN_WIN      TAOS_DEF_ERROR_CODE(0, 0x265E)
 #define TSDB_CODE_PAR_ONLY_SUPPORT_SINGLE_TABLE TAOS_DEF_ERROR_CODE(0, 0x265F)
-#define TSDB_CODE_PAR_INVALID_SMA_INDEX         TAOS_DEF_ERROR_CODE(0, 0x265C)
+#define TSDB_CODE_PAR_INVALID_SMA_INDEX         TAOS_DEF_ERROR_CODE(0, 0x2660)
+#define TSDB_CODE_PAR_INVALID_SELECTED_EXPR     TAOS_DEF_ERROR_CODE(0, 0x2661)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

source/common/src/ttime.c

@@ -710,6 +710,32 @@ int64_t taosTimeAdd(int64_t t, int64_t duration, char unit, int32_t precision) {
   return (int64_t)(taosMktime(&tm) * TSDB_TICK_PER_SECOND(precision));
 }
 
+int64_t taosTimeSub(int64_t t, int64_t duration, char unit, int32_t precision) {
+  if (duration == 0) {
+    return t;
+  }
+
+  if (unit != 'n' && unit != 'y') {
+    return t - duration;
+  }
+
+  // The following code handles the y/n time duration
+  int64_t numOfMonth = duration;
+  if (unit == 'y') {
+    numOfMonth *= 12;
+  }
+
+  struct tm tm;
+  time_t    tt = (time_t)(t / TSDB_TICK_PER_SECOND(precision));
+  taosLocalTime(&tt, &tm);
+  int32_t mon = tm.tm_year * 12 + tm.tm_mon - (int32_t)numOfMonth;
+  tm.tm_year = mon / 12;
+  tm.tm_mon = mon % 12;
+
+  return (int64_t)(taosMktime(&tm) * TSDB_TICK_PER_SECOND(precision));
+}
+
+
 int32_t taosTimeCountInterval(int64_t skey, int64_t ekey, int64_t interval, char unit, int32_t precision) {
   if (ekey < skey) {
     int64_t tmp = ekey;

source/libs/command/src/command.c

@@ -570,10 +570,14 @@ int32_t buildSelectResultDataBlock(SNodeList* pProjects, SSDataBlock* pBlock) {
   int32_t index = 0;
   SNode*  pProj = NULL;
   FOREACH(pProj, pProjects) {
-    if (((SValueNode*)pProj)->isNull) {
-      colDataAppend(taosArrayGet(pBlock->pDataBlock, index++), 0, NULL, true);
+    if (QUERY_NODE_VALUE != nodeType(pProj)) {
+      return TSDB_CODE_PAR_INVALID_SELECTED_EXPR;
     } else {
-      colDataAppend(taosArrayGet(pBlock->pDataBlock, index++), 0, nodesGetValueFromNode((SValueNode*)pProj), false);
+      if (((SValueNode*)pProj)->isNull) {
+        colDataAppend(taosArrayGet(pBlock->pDataBlock, index++), 0, NULL, true);
+      } else {
+        colDataAppend(taosArrayGet(pBlock->pDataBlock, index++), 0, nodesGetValueFromNode((SValueNode*)pProj), false);
+      }
     }
   }
 

source/libs/executor/src/sortoperator.c

@@ -621,7 +621,9 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
       break;
     }
   }
-
+  if (pInfo->groupSort) {
+    pInfo->hasGroupId = false;
+  }
   if (p->info.rows > 0) {  // todo extract method
     blockDataEnsureCapacity(pDataBlock, p->info.rows);
     int32_t numOfCols = taosArrayGetSize(pColMatchInfo);

source/libs/executor/src/timewindowoperator.c

@@ -1341,9 +1341,9 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
     uint64_t groupId = *(uint64_t*)key;
     ASSERT(keyLen == GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY)));
     TSKEY          ts = *(int64_t*)((char*)key + sizeof(uint64_t));
-    SResultRowInfo dumyInfo;
-    dumyInfo.cur.pageId = -1;
-    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, TSDB_ORDER_ASC);
+    STimeWindow win;
+    win.skey = ts;
+    win.ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
     SWinRes     winRe = {
             .ts = win.skey,
             .groupId = groupId,
@@ -4507,13 +4507,14 @@ static SSDataBlock* doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
       setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->order, scanFlag, true);
       doMergeAlignedIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);
       doFilter(miaInfo->pCondition, pRes);
-      if (pRes->info.rows > 0) {
+      if (pRes->info.rows >= pOperator->resultInfo.capacity) {
         break;
       }
     }
 
     pRes->info.groupId = miaInfo->groupId;
   }
+  miaInfo->hasGroupId = false;
 
   if (miaInfo->inputBlocksFinished) {
     doSetOperatorCompleted(pOperator);

source/libs/parser/src/parCalcConst.c

@@ -166,7 +166,7 @@ static int32_t calcConstStmtCondition(SCalcConstContext* pCxt, SNode** pCond, bo
   return code;
 }
 
-static int32_t calcConstProject(SNode* pProject, SNode** pNew) {
+static int32_t calcConstProject(SNode* pProject, bool dual, SNode** pNew) {
   SArray* pAssociation = NULL;
   if (NULL != ((SExprNode*)pProject)->pAssociation) {
     pAssociation = taosArrayDup(((SExprNode*)pProject)->pAssociation);
@@ -177,7 +177,12 @@ static int32_t calcConstProject(SNode* pProject, SNode** pNew) {
 
   char aliasName[TSDB_COL_NAME_LEN] = {0};
   strcpy(aliasName, ((SExprNode*)pProject)->aliasName);
-  int32_t code = scalarCalculateConstants(pProject, pNew);
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (dual) {
+    code = scalarCalculateConstantsFromDual(pProject, pNew);
+  } else {
+    code = scalarCalculateConstants(pProject, pNew);
+  }
   if (TSDB_CODE_SUCCESS == code && QUERY_NODE_VALUE == nodeType(*pNew) && NULL != pAssociation) {
     strcpy(((SExprNode*)*pNew)->aliasName, aliasName);
     int32_t size = taosArrayGetSize(pAssociation);
@@ -223,7 +228,7 @@ static int32_t calcConstProjections(SCalcConstContext* pCxt, SSelectStmt* pSelec
       continue;
     }
     SNode*  pNew = NULL;
-    int32_t code = calcConstProject(pProj, &pNew);
+    int32_t code = calcConstProject(pProj, (NULL == pSelect->pFromTable), &pNew);
     if (TSDB_CODE_SUCCESS == code) {
       REPLACE_NODE(pNew);
     } else {

source/libs/parser/src/parTranslater.c

@@ -916,8 +916,6 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
   }
 
   if (TSDB_DATA_TYPE_NULL == pVal->node.resType.type) {
-    // TODO
-    // pVal->node.resType = targetDt;
     pVal->translate = true;
     pVal->isNull = true;
     return DEAL_RES_CONTINUE;
@@ -932,6 +930,7 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
     res = translateNormalValue(pCxt, pVal, targetDt, strict);
   }
   pVal->node.resType = targetDt;
+  pVal->node.resType.scale = pVal->unit;
   pVal->translate = true;
   return res;
 }

source/libs/planner/src/planOptimizer.c

@@ -124,12 +124,15 @@ static bool scanPathOptMayBeOptimized(SLogicNode* pNode) {
                                  QUERY_NODE_LOGIC_PLAN_PARTITION != nodeType(pNode->pParent))) {
     return false;
   }
-  if (QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent) ||
+  if ((QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent) && WINDOW_TYPE_INTERVAL == ((SWindowLogicNode*)pNode->pParent)->winType) ||
       (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode->pParent) && pNode->pParent->pParent &&
-       QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent->pParent))) {
+       QUERY_NODE_LOGIC_PLAN_WINDOW == nodeType(pNode->pParent->pParent) && WINDOW_TYPE_INTERVAL == ((SWindowLogicNode*)pNode->pParent)->winType)) {
     return true;
   }
-  return !scanPathOptHaveNormalCol(((SAggLogicNode*)pNode->pParent)->pGroupKeys);
+  if (QUERY_NODE_LOGIC_PLAN_AGG == nodeType(pNode->pParent)) {
+    return !scanPathOptHaveNormalCol(((SAggLogicNode*)pNode->pParent)->pGroupKeys);
+  }
+  return false;
 }
 
 static SNodeList* scanPathOptGetAllFuncs(SLogicNode* pNode) {

source/libs/scalar/inc/sclInt.h

@@ -30,6 +30,7 @@ typedef struct SOperatorValueType {
 
 typedef struct SScalarCtx {
   int32_t      code;
+  bool         dual;
   SArray      *pBlockList;  /* element is SSDataBlock* */
   SHashObj    *pRes;  /* element is SScalarParam */
   void        *param;      // additional parameter (meta actually) for acquire value such as tbname/tags values

source/libs/scalar/src/scalar.c

@@ -1010,13 +1010,14 @@ int32_t sclExtendResRows(SScalarParam *pDst, SScalarParam *pSrc, SArray *pBlockL
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
+int32_t sclCalcConstants(SNode *pNode, bool dual, SNode **pRes) {
   if (NULL == pNode) {
     SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
 
   int32_t code = 0;
   SScalarCtx ctx = {0};
+  ctx.dual = dual;
   ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   if (NULL == ctx.pRes) {
     sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
@@ -1028,54 +1029,12 @@ int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
   *pRes = pNode;
 
 _return:
-  sclFreeRes(ctx.pRes);
-  return code;
-}
-
-int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
-  if (NULL == pNode || NULL == pBlockList) {
-    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
-  }
-
-  int32_t code = 0;
-  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList, .param = pDst ? pDst->param : NULL};
-
-  // TODO: OPT performance
-  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
-  if (NULL == ctx.pRes) {
-    sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
-    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
-  }
-  
-  nodesWalkExprPostOrder(pNode, sclCalcWalker, (void *)&ctx);
-  SCL_ERR_JRET(ctx.code);
-
-  if (pDst) {
-    SScalarParam *res = (SScalarParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
-    if (NULL == res) {
-      sclError("no valid res in hash, node:%p, type:%d", pNode, nodeType(pNode));
-      SCL_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
-    }
 
-    if (1 == res->numOfRows) {
-      SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList));
-    } else {
-      colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows);
-      colDataAssign(pDst->columnData, res->columnData, res->numOfRows, NULL);
-      pDst->numOfRows = res->numOfRows;
-    }
-
-    sclFreeParam(res);
-    taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
-  }
-
-_return:
-  //nodesDestroyNode(pNode);
   sclFreeRes(ctx.pRes);
   return code;
 }
 
-static int32_t getMinusOperatorResultType(SOperatorNode* pOp) {
+static int32_t sclGetMinusOperatorResType(SOperatorNode* pOp) {
   if (!IS_MATHABLE_TYPE(((SExprNode*)(pOp->pLeft))->resType.type)) {
     return TSDB_CODE_TSC_INVALID_OPERATION;
   }
@@ -1084,7 +1043,7 @@ static int32_t getMinusOperatorResultType(SOperatorNode* pOp) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t getArithmeticOperatorResultType(SOperatorNode* pOp) {
+static int32_t sclGetMathOperatorResType(SOperatorNode* pOp) {
   SDataType ldt = ((SExprNode*)(pOp->pLeft))->resType;
   SDataType rdt = ((SExprNode*)(pOp->pRight))->resType;
   if ((TSDB_DATA_TYPE_TIMESTAMP == ldt.type && TSDB_DATA_TYPE_TIMESTAMP == rdt.type) ||
@@ -1106,7 +1065,7 @@ static int32_t getArithmeticOperatorResultType(SOperatorNode* pOp) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t getComparisonOperatorResultType(SOperatorNode* pOp) {
+static int32_t sclGetCompOperatorResType(SOperatorNode* pOp) {
   SDataType ldt = ((SExprNode*)(pOp->pLeft))->resType;
   if (OP_TYPE_IN == pOp->opType || OP_TYPE_NOT_IN == pOp->opType) {
     ((SExprNode*)(pOp->pRight))->resType = ldt;
@@ -1122,7 +1081,7 @@ static int32_t getComparisonOperatorResultType(SOperatorNode* pOp) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t getJsonOperatorResultType(SOperatorNode* pOp) {
+static int32_t sclGetJsonOperatorResType(SOperatorNode* pOp) {
   SDataType ldt = ((SExprNode*)(pOp->pLeft))->resType;
   SDataType rdt = ((SExprNode*)(pOp->pRight))->resType;
   if (TSDB_DATA_TYPE_JSON != ldt.type || !IS_STR_DATA_TYPE(rdt.type)) {
@@ -1137,12 +1096,64 @@ static int32_t getJsonOperatorResultType(SOperatorNode* pOp) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t getBitwiseOperatorResultType(SOperatorNode* pOp) {
+static int32_t sclGetBitwiseOperatorResType(SOperatorNode* pOp) {
   pOp->node.resType.type = TSDB_DATA_TYPE_BIGINT;
   pOp->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes;
   return TSDB_CODE_SUCCESS;
 }
 
+
+int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
+  return sclCalcConstants(pNode, false, pRes);
+}
+
+int32_t scalarCalculateConstantsFromDual(SNode *pNode, SNode **pRes) {
+  return sclCalcConstants(pNode, true, pRes);
+}
+
+int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
+  if (NULL == pNode || NULL == pBlockList) {
+    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
+  }
+
+  int32_t code = 0;
+  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList, .param = pDst ? pDst->param : NULL};
+
+  // TODO: OPT performance
+  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
+  if (NULL == ctx.pRes) {
+    sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
+    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
+  }
+  
+  nodesWalkExprPostOrder(pNode, sclCalcWalker, (void *)&ctx);
+  SCL_ERR_JRET(ctx.code);
+
+  if (pDst) {
+    SScalarParam *res = (SScalarParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
+    if (NULL == res) {
+      sclError("no valid res in hash, node:%p, type:%d", pNode, nodeType(pNode));
+      SCL_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
+    }
+
+    if (1 == res->numOfRows) {
+      SCL_ERR_JRET(sclExtendResRows(pDst, res, pBlockList));
+    } else {
+      colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows);
+      colDataAssign(pDst->columnData, res->columnData, res->numOfRows, NULL);
+      pDst->numOfRows = res->numOfRows;
+    }
+
+    sclFreeParam(res);
+    taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
+  }
+
+_return:
+  //nodesDestroyNode(pNode);
+  sclFreeRes(ctx.pRes);
+  return code;
+}
+
 int32_t scalarGetOperatorResultType(SOperatorNode* pOp) {
   if (TSDB_DATA_TYPE_BLOB == ((SExprNode*)(pOp->pLeft))->resType.type ||
       (NULL != pOp->pRight && TSDB_DATA_TYPE_BLOB == ((SExprNode*)(pOp->pRight))->resType.type)) {
@@ -1155,15 +1166,15 @@ int32_t scalarGetOperatorResultType(SOperatorNode* pOp) {
     case OP_TYPE_MULTI:
     case OP_TYPE_DIV:
     case OP_TYPE_REM:
-      return getArithmeticOperatorResultType(pOp);
+      return sclGetMathOperatorResType(pOp);
     case OP_TYPE_MINUS:
-      return getMinusOperatorResultType(pOp);
+      return sclGetMinusOperatorResType(pOp);
     case OP_TYPE_ASSIGN:
       pOp->node.resType = ((SExprNode*)(pOp->pLeft))->resType;
       break;
     case OP_TYPE_BIT_AND:
     case OP_TYPE_BIT_OR:
-      return getBitwiseOperatorResultType(pOp);
+      return sclGetBitwiseOperatorResType(pOp);
     case OP_TYPE_GREATER_THAN:
     case OP_TYPE_GREATER_EQUAL:
     case OP_TYPE_LOWER_THAN:
@@ -1184,10 +1195,10 @@ int32_t scalarGetOperatorResultType(SOperatorNode* pOp) {
     case OP_TYPE_NMATCH:
     case OP_TYPE_IN:
     case OP_TYPE_NOT_IN:
-      return getComparisonOperatorResultType(pOp);
+      return sclGetCompOperatorResType(pOp);
     case OP_TYPE_JSON_GET_VALUE:
     case OP_TYPE_JSON_CONTAINS:
-      return getJsonOperatorResultType(pOp);
+      return sclGetJsonOperatorResType(pOp);
     default:
       break;
   }

source/libs/scalar/src/sclvector.c

@@ -1055,7 +1055,7 @@ static void vectorMathAddHelper(SColumnInfoData* pLeftCol, SColumnInfoData* pRig
   }
 }
 
-static void vectorMathBigintAddHelper(SColumnInfoData* pLeftCol, SColumnInfoData* pRightCol, SColumnInfoData* pOutputCol, int32_t numOfRows, int32_t step, int32_t i) {
+static void vectorMathTsAddHelper(SColumnInfoData* pLeftCol, SColumnInfoData* pRightCol, SColumnInfoData* pOutputCol, int32_t numOfRows, int32_t step, int32_t i) {
   _getBigintValue_fn_t getVectorBigintValueFnLeft  = getVectorBigintValueFn(pLeftCol->info.type);
   _getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
 
@@ -1069,7 +1069,8 @@ static void vectorMathBigintAddHelper(SColumnInfoData* pLeftCol, SColumnInfoData
         colDataAppendNULL(pOutputCol, i);
         continue;  // TODO set null or ignore
       }
-      *output = getVectorBigintValueFnLeft(pLeftCol->pData, i) + getVectorBigintValueFnRight(pRightCol->pData, 0);
+      *output = taosTimeAdd(getVectorBigintValueFnLeft(pLeftCol->pData, i), getVectorBigintValueFnRight(pRightCol->pData, 0),
+                            pRightCol->info.scale, pRightCol->info.precision);
     }
   }
 }
@@ -1116,7 +1117,17 @@ void vectorMathAdd(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut
     _getBigintValue_fn_t getVectorBigintValueFnLeft  = getVectorBigintValueFn(pLeftCol->info.type);
     _getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
 
-    if (pLeft->numOfRows == pRight->numOfRows) {
+    if (pLeft->numOfRows == 1 && pRight->numOfRows == 1) {
+      if (GET_PARAM_TYPE(pLeft) == TSDB_DATA_TYPE_TIMESTAMP) {
+        vectorMathTsAddHelper(pLeftCol, pRightCol, pOutputCol, pRight->numOfRows, step, i);
+      } else {
+        vectorMathTsAddHelper(pRightCol, pLeftCol, pOutputCol, pRight->numOfRows, step, i);
+      }
+    } else if (pLeft->numOfRows == 1) {
+      vectorMathTsAddHelper(pRightCol, pLeftCol, pOutputCol, pRight->numOfRows, step, i);
+    } else if (pRight->numOfRows == 1) {
+      vectorMathTsAddHelper(pLeftCol, pRightCol, pOutputCol, pLeft->numOfRows, step, i);
+    } else if (pLeft->numOfRows == pRight->numOfRows) {
       for (; i < pRight->numOfRows && i >= 0; i += step, output += 1) {
         if (IS_NULL) {
           colDataAppendNULL(pOutputCol, i);
@@ -1124,11 +1135,7 @@ void vectorMathAdd(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut
         }
         *output = getVectorBigintValueFnLeft(pLeftCol->pData, i) + getVectorBigintValueFnRight(pRightCol->pData, i);
       }
-    } else if (pLeft->numOfRows == 1) {
-      vectorMathBigintAddHelper(pRightCol, pLeftCol, pOutputCol, pRight->numOfRows, step, i);
-    } else if (pRight->numOfRows == 1) {
-      vectorMathBigintAddHelper(pLeftCol, pRightCol, pOutputCol, pLeft->numOfRows, step, i);
-    }
+    } 
   } else {
     double *output = (double *)pOutputCol->pData;
     _getDoubleValue_fn_t getVectorDoubleValueFnLeft  = getVectorDoubleValueFn(pLeftCol->info.type);
@@ -1174,7 +1181,7 @@ static void vectorMathSubHelper(SColumnInfoData* pLeftCol, SColumnInfoData* pRig
   }
 }
 
-static void vectorMathBigintSubHelper(SColumnInfoData* pLeftCol, SColumnInfoData* pRightCol, SColumnInfoData* pOutputCol, int32_t numOfRows, int32_t step, int32_t factor, int32_t i) {
+static void vectorMathTsSubHelper(SColumnInfoData* pLeftCol, SColumnInfoData* pRightCol, SColumnInfoData* pOutputCol, int32_t numOfRows, int32_t step, int32_t factor, int32_t i) {
   _getBigintValue_fn_t getVectorBigintValueFnLeft  = getVectorBigintValueFn(pLeftCol->info.type);
   _getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
 
@@ -1188,7 +1195,9 @@ static void vectorMathBigintSubHelper(SColumnInfoData* pLeftCol, SColumnInfoData
         colDataAppendNULL(pOutputCol, i);
         continue;  // TODO set null or ignore
       }
-      *output = (getVectorBigintValueFnLeft(pLeftCol->pData, i) - getVectorBigintValueFnRight(pRightCol->pData, 0)) * factor;
+      *output = taosTimeSub(getVectorBigintValueFnLeft(pLeftCol->pData, i), getVectorBigintValueFnRight(pRightCol->pData, 0),
+                            pRightCol->info.scale, pRightCol->info.precision);
+      
     }
   }
 }
@@ -1211,7 +1220,13 @@ void vectorMathSub(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut
     _getBigintValue_fn_t getVectorBigintValueFnLeft  = getVectorBigintValueFn(pLeftCol->info.type);
     _getBigintValue_fn_t getVectorBigintValueFnRight = getVectorBigintValueFn(pRightCol->info.type);
 
-    if (pLeft->numOfRows == pRight->numOfRows) {
+    if (pLeft->numOfRows == 1 && pRight->numOfRows == 1) {
+      vectorMathTsSubHelper(pLeftCol, pRightCol, pOutputCol, pLeft->numOfRows, step, 1, i);
+    } else if (pLeft->numOfRows == 1) {
+      vectorMathTsSubHelper(pRightCol, pLeftCol, pOutputCol, pRight->numOfRows, step, -1, i);
+    } else if (pRight->numOfRows == 1) {
+      vectorMathTsSubHelper(pLeftCol, pRightCol, pOutputCol, pLeft->numOfRows, step, 1, i);
+    } else if (pLeft->numOfRows == pRight->numOfRows) {
       for (; i < pRight->numOfRows && i >= 0; i += step, output += 1) {
         if (IS_NULL) {
           colDataAppendNULL(pOutputCol, i);
@@ -1219,10 +1234,6 @@ void vectorMathSub(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut
         }
         *output = getVectorBigintValueFnLeft(pLeftCol->pData, i) - getVectorBigintValueFnRight(pRightCol->pData, i);
       }
-    } else if (pLeft->numOfRows == 1) {
-      vectorMathBigintSubHelper(pRightCol, pLeftCol, pOutputCol, pRight->numOfRows, step, -1, i);
-    } else if (pRight->numOfRows == 1) {
-      vectorMathBigintSubHelper(pLeftCol, pRightCol, pOutputCol, pLeft->numOfRows, step, 1, i);
     }
   } else {
     double *output = (double *)pOutputCol->pData;

source/util/src/terror.c

@@ -559,6 +559,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_COL_JSON,           "Only tag can be jso
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_VALUE_TOO_LONG,             "Value too long for column/tag")
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_DELETE_WHERE,       "The DELETE statement must have a definite time window range")
 TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_REDISTRIBUTE_VG,    "The REDISTRIBUTE VGROUP statement only support 1 to 3 dnodes")
+TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_SELECTED_EXPR,      "Invalid SELECTed expression")
 
 //planner
 TAOS_DEFINE_ERROR(TSDB_CODE_PLAN_INTERNAL_ERROR,            "Planner internal error")

tests/script/jenkins/basic.txt

@@ -224,5 +224,6 @@
 
 # --- scalar
 ./test.sh -f tsim/scalar/in.sim
+./test.sh -f tsim/scalar/scalar.sim
 
 #======================b1-end===============

tests/script/tsim/scalar/scalar.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sql connect
+
+print ======== step1 
+sql drop database if exists db1;
+sql create database db1 vgroups 3;
+sql use db1;
+sql create stable st1 (fts timestamp, fbool bool, ftiny tinyint, fsmall smallint, fint int, fbig bigint, futiny tinyint unsigned, fusmall smallint unsigned, fuint int unsigned, fubig bigint unsigned, ffloat float, fdouble double, fbin binary(10), fnchar nchar(10)) tags(tts timestamp, tbool bool, ttiny tinyint, tsmall smallint, tint int, tbig bigint, tutiny tinyint unsigned, tusmall smallint unsigned, tuint int unsigned, tubig bigint unsigned, tfloat float, tdouble double, tbin binary(10), tnchar nchar(10));
+sql create table tb1 using st1 tags('2022-07-10 16:31:00', true, 1, 1, 1, 1, 1, 1, 1, 1, 1.0, 1.0, 'a', 'a');
+sql create table tb2 using st1 tags('2022-07-10 16:32:00', false, 2, 2, 2, 2, 2, 2, 2, 2, 2.0, 2.0, 'b', 'b');
+sql create table tb3 using st1 tags('2022-07-10 16:33:00', true, 3, 3, 3, 3, 3, 3, 3, 3, 3.0, 3.0, 'c', 'c');
+ 
+sql insert into tb1 values ('2022-07-10 16:31:01', false, 1, 1, 1, 1, 1, 1, 1, 1, 1.0, 1.0, 'a', 'a');
+sql insert into tb1 values ('2022-07-10 16:31:02', true, 2, 2, 2, 2, 2, 2, 2, 2, 2.0, 2.0, 'b', 'b');
+sql insert into tb1 values ('2022-07-10 16:31:03', false, 3, 3, 3, 3, 3, 3, 3, 3, 3.0, 3.0, 'c', 'c');
+sql insert into tb1 values ('2022-07-10 16:31:04', true, 4, 4, 4, 4, 4, 4, 4, 4, 4.0, 4.0, 'd', 'd');
+sql insert into tb1 values ('2022-07-10 16:31:05', false, 5, 5, 5, 5, 5, 5, 5, 5, 5.0, 5.0, 'e', 'e');
+ 
+sql insert into tb2 values ('2022-07-10 16:32:01', false, 1, 1, 1, 1, 1, 1, 1, 1, 1.0, 1.0, 'a', 'a');
+sql insert into tb2 values ('2022-07-10 16:32:02', true, 2, 2, 2, 2, 2, 2, 2, 2, 2.0, 2.0, 'b', 'b'); 
+sql insert into tb2 values ('2022-07-10 16:32:03', false, 3, 3, 3, 3, 3, 3, 3, 3, 3.0, 3.0, 'c', 'c');
+sql insert into tb2 values ('2022-07-10 16:32:04', true, 4, 4, 4, 4, 4, 4, 4, 4, 4.0, 4.0, 'd', 'd'); 
+sql insert into tb2 values ('2022-07-10 16:32:05', false, 5, 5, 5, 5, 5, 5, 5, 5, 5.0, 5.0, 'e', 'e');
+ 
+sql insert into tb3 values ('2022-07-10 16:33:01', false, 1, 1, 1, 1, 1, 1, 1, 1, 1.0, 1.0, 'a', 'a');
+sql insert into tb3 values ('2022-07-10 16:33:02', true, 2, 2, 2, 2, 2, 2, 2, 2, 2.0, 2.0, 'b', 'b'); 
+sql insert into tb3 values ('2022-07-10 16:33:03', false, 3, 3, 3, 3, 3, 3, 3, 3, 3.0, 3.0, 'c', 'c');
+sql insert into tb3 values ('2022-07-10 16:33:04', true, 4, 4, 4, 4, 4, 4, 4, 4, 4.0, 4.0, 'd', 'd'); 
+sql insert into tb3 values ('2022-07-10 16:33:05', false, 5, 5, 5, 5, 5, 5, 5, 5, 5.0, 5.0, 'e', 'e'); 
+
+sql select 1+1n;
+if $rows != 1 then
+  return -1
+endi
+if $data00 != 2.000000000 then
+  return -1
+endi
+
+
+sql select cast(1 as timestamp)+1n;
+if $rows != 1 then
+  return -1
+endi
+if $data00 != @70-02-01 08:00:00.000@ then
+  return -1
+endi
+
+sql select 1-1n;
+if $rows != 1 then
+  return -1
+endi
+
+sql select cast(1 as timestamp)-1y;
+if $rows != 1 then
+  return -1
+endi
+if $data00 != @69-01-01 08:00:00.000@ then
+  return -1
+endi
+
+sql select 1n-now();
+sql select 1n+now();
+#sql select avg(4n);
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/stream/sliding.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+print =============== create database
+sql create database test vgroups 1
+sql show databases
+if $rows != 3 then
+  return -1
+endi
+
+print $data00 $data01 $data02
+
+sql use test
+sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
+sql create table t1 using st tags(1,1,1);
+sql create table t2 using st tags(2,2,2);
+sql create stream streams1 trigger at_once  into streamt as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from t1 interval(10s) sliding (5s);
+sql create stream streams2 trigger at_once  watermark 1d into streamt2 as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from t1 interval(10s) sliding (5s);
+sql create stream stream_t1 trigger at_once  into streamtST as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from st interval(10s) sliding (5s);
+sql create stream stream_t2 trigger at_once watermark 1d into streamtST2 as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from st interval(10s) sliding (5s);
+
+sql insert into t1 values(1648791210000,1,2,3,1.0);
+sql insert into t1 values(1648791216000,2,2,3,1.1);
+sql insert into t1 values(1648791220000,3,2,3,2.1);
+
+sql insert into t1 values(1648791210000,1,2,3,1.0);
+sql insert into t1 values(1648791216000,2,2,3,1.1);
+sql insert into t1 values(1648791220000,3,2,3,2.1);
+
+sql insert into t2 values(1648791210000,1,2,3,1.0);
+sql insert into t2 values(1648791216000,2,2,3,1.1);
+sql insert into t2 values(1648791220000,3,2,3,2.1);
+
+sql insert into t2 values(1648791210000,1,2,3,1.0);
+sql insert into t2 values(1648791216000,2,2,3,1.1);
+sql insert into t2 values(1648791220000,3,2,3,2.1);
+
+$loop_count = 0
+
+loop0:
+sleep 300
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+
+sql select * from streamt
+
+# row 0
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop0
+endi
+
+# row 1
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop0
+endi
+
+if $data12 != 3 then
+  print =====data12=$data12
+  goto loop0
+endi
+
+# row 2
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop0
+endi
+
+if $data22 != 5 then
+  print =====data22=$data22
+  goto loop0
+endi
+
+# row 3
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop0
+endi
+
+if $data32 != 3 then
+  print =====data32=$data32
+  goto loop0
+endi
+
+print step 1
+
+sql select * from streamt2
+
+# row 0
+if $data01 != 1 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != 1 then
+  print =====data02=$data02
+  goto loop0
+endi
+
+# row 1
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop0
+endi
+
+if $data12 != 3 then
+  print =====data12=$data12
+  goto loop0
+endi
+
+# row 2
+if $data21 != 2 then
+  print =====data21=$data21
+  goto loop0
+endi
+
+if $data22 != 5 then
+  print =====data22=$data22
+  goto loop0
+endi
+
+# row 3
+if $data31 != 1 then
+  print =====data31=$data31
+  goto loop0
+endi
+
+if $data32 != 3 then
+  print =====data32=$data32
+  goto loop0
+endi
+
+print step 2
+
+sql select * from streamtST
+
+# row 0
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop0
+endi
+
+# row 1
+if $data11 != 4 then
+  print =====data11=$data11
+  goto loop0
+endi
+
+if $data12 != 6 then
+  print =====data12=$data12
+  goto loop0
+endi
+
+# row 2
+if $data21 != 4 then
+  print =====data21=$data21
+  goto loop0
+endi
+
+if $data22 != 10 then
+  print =====data22=$data22
+  goto loop0
+endi
+
+# row 3
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop0
+endi
+
+if $data32 != 6 then
+  print =====data32=$data32
+  goto loop0
+endi
+
+print step 3
+
+sql select * from streamtST2
+
+# row 0
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop0
+endi
+
+if $data02 != 2 then
+  print =====data02=$data02
+  goto loop0
+endi
+
+# row 1
+if $data11 != 4 then
+  print =====data11=$data11
+  goto loop0
+endi
+
+if $data12 != 6 then
+  print =====data12=$data12
+  goto loop0
+endi
+
+# row 2
+if $data21 != 4 then
+  print =====data21=$data21
+  goto loop0
+endi
+
+if $data22 != 10 then
+  print =====data22=$data22
+  goto loop0
+endi
+
+# row 3
+if $data31 != 2 then
+  print =====data31=$data31
+  goto loop0
+endi
+
+if $data32 != 6 then
+  print =====data32=$data32
+  goto loop0
+endi
+
+
+system sh/stop_dnodes.sh
\ No newline at end of file

tests/system-test/7-tmq/tmqUpdate-1ctb.py

@@ -19,7 +19,7 @@ class TDTestCase:
         self.snapshot   = 0
         self.vgroups    = 4
         self.ctbNum     = 1
-        self.rowsPerTbl = 100000
+        self.rowsPerTbl = 10000
         
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")

tests/system-test/fulltest.sh

@@ -148,7 +148,7 @@ python3 ./test.py -f 7-tmq/subscribeDb2.py
 python3 ./test.py -f 7-tmq/subscribeDb3.py
 #python3 ./test.py -f 7-tmq/subscribeDb4.py
 python3 ./test.py -f 7-tmq/subscribeStb.py
-#python3 ./test.py -f 7-tmq/subscribeStb0.py
+python3 ./test.py -f 7-tmq/subscribeStb0.py
 python3 ./test.py -f 7-tmq/subscribeStb1.py
 python3 ./test.py -f 7-tmq/subscribeStb2.py
 python3 ./test.py -f 7-tmq/subscribeStb3.py