other: merge 3.0

source/libs/executor/inc/executorimpl.h

@@ -153,17 +153,17 @@ typedef struct {
   SSchemaWrapper* qsw;
 } SSchemaInfo;
 
-typedef struct {
+typedef struct SExchangeOpStopInfo {
   int32_t operatorType;
   int64_t refId;
 } SExchangeOpStopInfo;
 
-typedef struct {
+typedef struct STaskStopInfo {
   SRWLatch lock;
   SArray*  pStopInfo;
 } STaskStopInfo;
 
-typedef struct SExecTaskInfo {
+struct SExecTaskInfo {
   STaskIdInfo   id;
   uint32_t      status;
   STimeWindow   window;
@@ -260,7 +260,7 @@ typedef struct SExchangeInfo {
 
   // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
   // passed by downstream operator
-  SArray*             pReadyBlocks;
+  SArray*             pResultBlockList;
   SArray*             pRecycledBlocks;// build a pool for small data block to avoid to repeatly create and then destroy.
   SSDataBlock*        pDummyBlock;    // dummy block, not keep data
   bool                seqLoadData;    // sequential load data or not, false by default

source/libs/executor/src/exchangeoperator.c

@@ -182,7 +182,7 @@ static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
   }
 
   // we have buffered retrieved datablock, return it directly
-  SSDataBlock** p = taosArrayPop(pExchangeInfo->pReadyBlocks);
+  SSDataBlock** p = taosArrayPop(pExchangeInfo->pResultBlockList);
   if (p != NULL) {
     taosArrayPush(pExchangeInfo->pRecycledBlocks, p);
     return *p;
@@ -193,10 +193,10 @@ static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
       concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
     }
 
-    if (taosArrayGetSize(pExchangeInfo->pReadyBlocks) == 0) {
+    if (taosArrayGetSize(pExchangeInfo->pResultBlockList) == 0) {
       return NULL;
     } else {
-      p = taosArrayPop(pExchangeInfo->pReadyBlocks);
+      p = taosArrayPop(pExchangeInfo->pResultBlockList);
       taosArrayPush(pExchangeInfo->pRecycledBlocks, p);
       return *p;
     }
@@ -298,7 +298,7 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
 
   tsem_init(&pInfo->ready, 0, 0);
   pInfo->pDummyBlock = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
-  pInfo->pReadyBlocks = taosArrayInit(64, POINTER_BYTES);
+  pInfo->pResultBlockList = taosArrayInit(64, POINTER_BYTES);
   pInfo->pRecycledBlocks = taosArrayInit(64, POINTER_BYTES);
 
   SExchangeOpStopInfo stopInfo = {QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, pInfo->self};
@@ -346,7 +346,7 @@ void doDestroyExchangeOperatorInfo(void* param) {
   taosArrayDestroy(pExInfo->pSources);
   taosArrayDestroyEx(pExInfo->pSourceDataInfo, freeSourceDataInfo);
 
-  taosArrayDestroyEx(pExInfo->pReadyBlocks, freeBlock);
+  taosArrayDestroyEx(pExInfo->pResultBlockList, freeBlock);
   taosArrayDestroyEx(pExInfo->pRecycledBlocks, freeBlock);
 
   blockDataDestroy(pExInfo->pDummyBlock);

source/libs/function/inc/builtinsimpl.h

@@ -30,6 +30,8 @@ typedef struct SSumRes {
     double   dsum;
   };
   int16_t type;
+  int64_t prevTs;
+  bool    isPrevTsSet;
 } SSumRes;
 
 typedef struct SMinmaxResInfo {