diff --git a/include/libs/executor/dataSinkMgt.h b/include/libs/executor/dataSinkMgt.h
index 5cef3b2253..371cb12405 100644
--- a/include/libs/executor/dataSinkMgt.h
+++ b/include/libs/executor/dataSinkMgt.h
@@ -70,7 +70,7 @@ int32_t dsCreateDataSinker(const struct SDataSink *pDataSink, DataSinkHandle* pH
  */
 int32_t dsPutDataBlock(DataSinkHandle handle, const SInputData* pInput, bool* pContinue);
 
-void dsEndPut(DataSinkHandle handle, int64_t useconds);
+void dsEndPut(DataSinkHandle handle, uint64_t useconds);
 
 /**
  * Get the length of the data returned by the next call to dsGetDataBlock.
diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h
index 71b014d025..0fc7fd679e 100644
--- a/include/libs/executor/executor.h
+++ b/include/libs/executor/executor.h
@@ -20,6 +20,8 @@
 extern "C" {
 #endif
 
+#include "common.h"
+
 typedef void* qTaskInfo_t;
 typedef void* DataSinkHandle;
 struct SSubplan;
@@ -34,7 +36,7 @@ struct SSubplan;
   * @param qId
   * @return
   */
-int32_t qCreateExecTask(void* tsdb, int32_t vgId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo);
+int32_t qCreateExecTask(void* tsdb, int32_t vgId, struct SSubplan* pPlan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle);
 
 /**
  * The main task execution function, including query on both table and multiple tables,
@@ -44,7 +46,7 @@ int32_t qCreateExecTask(void* tsdb, int32_t vgId, struct SSubplan* pPlan, qTaskI
  * @param handle
  * @return
  */
-int32_t qExecTask(qTaskInfo_t tinfo, DataSinkHandle* handle);
+int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds);
 
 /**
  * Retrieve the produced results information, if current query is not paused or completed,
diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c
index bf80490416..d07a5ffc77 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead.c
@@ -3454,6 +3454,7 @@ void filterPrepare(void* expr, void* param) {
   }
 }
 
+
 static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
   STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
   STable* pTable1 = ((STableKeyInfo*) p1)->pTable;
@@ -3537,8 +3538,6 @@ void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTable
     int32_t ret = compareFn(prev, p, pSupp);
     assert(ret == 0 || ret == -1);
 
-//    assert((*p)->type == TSDB_CHILD_TABLE);
-
     if (ret == 0) {
       STableKeyInfo info1 = {.pTable = *p, .lastKey = skey};
       taosArrayPush(g, &info1);
@@ -3554,7 +3553,6 @@ void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTable
   taosArrayPush(pGroups, &g);
 }
 
-#if 0
 SArray* createTableGroup(SArray* pTableList, STSchema* pTagSchema, SColIndex* pCols, int32_t numOfOrderCols, TSKEY skey) {
   assert(pTableList != NULL);
   SArray* pTableGroup = taosArrayInit(1, POINTER_BYTES);
@@ -3587,145 +3585,138 @@ SArray* createTableGroup(SArray* pTableList, STSchema* pTagSchema, SColIndex* pC
     sup.pTagSchema = pTagSchema;
     sup.pCols = pCols;
 
-    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
-    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
+//    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
+//    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
   }
 
   return pTableGroup;
 }
 
-static bool tableFilterFp(const void* pNode, void* param) {
-  tQueryInfo* pInfo = (tQueryInfo*) param;
+//static bool tableFilterFp(const void* pNode, void* param) {
+//  tQueryInfo* pInfo = (tQueryInfo*) param;
+//
+//  STable* pTable = (STable*)(SL_GET_NODE_DATA((SSkipListNode*)pNode));
+//
+//  char* val = NULL;
+//  if (pInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
+//    val = (char*) TABLE_NAME(pTable);
+//  } else {
+//    val = tdGetKVRowValOfCol(pTable->tagVal, pInfo->sch.colId);
+//  }
+//
+//  if (pInfo->optr == TSDB_RELATION_ISNULL || pInfo->optr == TSDB_RELATION_NOTNULL) {
+//    if (pInfo->optr == TSDB_RELATION_ISNULL) {
+//      return (val == NULL) || isNull(val, pInfo->sch.type);
+//    } else if (pInfo->optr == TSDB_RELATION_NOTNULL) {
+//      return (val != NULL) && (!isNull(val, pInfo->sch.type));
+//    }
+//  } else if (pInfo->optr == TSDB_RELATION_IN) {
+//     int type = pInfo->sch.type;
+//     if (type == TSDB_DATA_TYPE_BOOL || IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_TIMESTAMP) {
+//       int64_t v;
+//       GET_TYPED_DATA(v, int64_t, pInfo->sch.type, val);
+//       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));
+//     } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
+//       uint64_t v;
+//       GET_TYPED_DATA(v, uint64_t, pInfo->sch.type, val);
+//       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));
+//     }
+//     else if (type == TSDB_DATA_TYPE_DOUBLE || type == TSDB_DATA_TYPE_FLOAT) {
+//       double v;
+//       GET_TYPED_DATA(v, double, pInfo->sch.type, val);
+//       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));
+//     } else if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR){
+//       return NULL != taosHashGet((SHashObj *)pInfo->q, varDataVal(val), varDataLen(val));
+//     }
+//
+//  }
+//
+//  int32_t ret = 0;
+//  if (val == NULL) { //the val is possible to be null, so check it out carefully
+//    ret = -1; // val is missing in table tags value pairs
+//  } else {
+//    ret = pInfo->compare(val, pInfo->q);
+//  }
+//
+//  switch (pInfo->optr) {
+//    case TSDB_RELATION_EQUAL: {
+//      return ret == 0;
+//    }
+//    case TSDB_RELATION_NOT_EQUAL: {
+//      return ret != 0;
+//    }
+//    case TSDB_RELATION_GREATER_EQUAL: {
+//      return ret >= 0;
+//    }
+//    case TSDB_RELATION_GREATER: {
+//      return ret > 0;
+//    }
+//    case TSDB_RELATION_LESS_EQUAL: {
+//      return ret <= 0;
+//    }
+//    case TSDB_RELATION_LESS: {
+//      return ret < 0;
+//    }
+//    case TSDB_RELATION_LIKE: {
+//      return ret == 0;
+//    }
+//    case TSDB_RELATION_MATCH: {
+//      return ret == 0;
+//    }
+//    case TSDB_RELATION_NMATCH: {
+//      return ret == 0;
+//    }
+//    case TSDB_RELATION_IN: {
+//      return ret == 1;
+//    }
+//
+//    default:
+//      assert(false);
+//  }
+//
+//  return true;
+//}
 
-  STable* pTable = (STable*)(SL_GET_NODE_DATA((SSkipListNode*)pNode));
+//static void getTableListfromSkipList(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp *param);
 
-  char* val = NULL;
-  if (pInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
-    val = (char*) TABLE_NAME(pTable);
-  } else {
-    val = tdGetKVRowValOfCol(pTable->tagVal, pInfo->sch.colId);
-  }
-
-  if (pInfo->optr == TSDB_RELATION_ISNULL || pInfo->optr == TSDB_RELATION_NOTNULL) {
-    if (pInfo->optr == TSDB_RELATION_ISNULL) {
-      return (val == NULL) || isNull(val, pInfo->sch.type);
-    } else if (pInfo->optr == TSDB_RELATION_NOTNULL) {
-      return (val != NULL) && (!isNull(val, pInfo->sch.type));
-    }
-  } else if (pInfo->optr == TSDB_RELATION_IN) {
-     int type = pInfo->sch.type;
-     if (type == TSDB_DATA_TYPE_BOOL || IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_TIMESTAMP) {
-       int64_t v;
-       GET_TYPED_DATA(v, int64_t, pInfo->sch.type, val);
-       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));     
-     } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
-       uint64_t v;
-       GET_TYPED_DATA(v, uint64_t, pInfo->sch.type, val);
-       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));     
-     }
-     else if (type == TSDB_DATA_TYPE_DOUBLE || type == TSDB_DATA_TYPE_FLOAT) {
-       double v;
-       GET_TYPED_DATA(v, double, pInfo->sch.type, val);
-       return NULL != taosHashGet((SHashObj *)pInfo->q, (char *)&v, sizeof(v));     
-     } else if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR){
-       return NULL != taosHashGet((SHashObj *)pInfo->q, varDataVal(val), varDataLen(val));        
-     }     
-
-  }
-
-  int32_t ret = 0;
-  if (val == NULL) { //the val is possible to be null, so check it out carefully
-    ret = -1; // val is missing in table tags value pairs
-  } else {
-    ret = pInfo->compare(val, pInfo->q);
-  }
-
-  switch (pInfo->optr) {
-    case TSDB_RELATION_EQUAL: {
-      return ret == 0;
-    }
-    case TSDB_RELATION_NOT_EQUAL: {
-      return ret != 0;
-    }
-    case TSDB_RELATION_GREATER_EQUAL: {
-      return ret >= 0;
-    }
-    case TSDB_RELATION_GREATER: {
-      return ret > 0;
-    }
-    case TSDB_RELATION_LESS_EQUAL: {
-      return ret <= 0;
-    }
-    case TSDB_RELATION_LESS: {
-      return ret < 0;
-    }
-    case TSDB_RELATION_LIKE: {
-      return ret == 0;
-    }
-    case TSDB_RELATION_MATCH: {
-      return ret == 0;
-    }
-    case TSDB_RELATION_NMATCH: {
-      return ret == 0;
-    }
-    case TSDB_RELATION_IN: {
-      return ret == 1;
-    }
-
-    default:
-      assert(false);
-  }
-
-  return true;
-}
-
-static void getTableListfromSkipList(tExprNode *pExpr, SSkipList *pSkipList, SArray *result, SExprTraverseSupp *param);
-
-static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr) {
-  // query according to the expression tree
-  SExprTraverseSupp supp = {
-      .nodeFilterFn = (__result_filter_fn_t) tableFilterFp,
-      .setupInfoFn = filterPrepare,
-      .pExtInfo = pSTable->tagSchema,
-      };
-
-  getTableListfromSkipList(pExpr, pSTable->pIndex, pRes, &supp);
-  tExprTreeDestroy(pExpr, destroyHelper);
-  return TSDB_CODE_SUCCESS;
-}
+//static int32_t doQueryTableList(STable* pSTable, SArray* pRes, tExprNode* pExpr) {
+//  // query according to the expression tree
+//  SExprTraverseSupp supp = {
+//      .nodeFilterFn = (__result_filter_fn_t) tableFilterFp,
+//      .setupInfoFn = filterPrepare,
+//      .pExtInfo = pSTable->tagSchema,
+//      };
+//
+//  getTableListfromSkipList(pExpr, pSTable->pIndex, pRes, &supp);
+//  tExprTreeDestroy(pExpr, destroyHelper);
+//  return TSDB_CODE_SUCCESS;
+//}
 
 int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len,
                                  int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo,
-                                 SColIndex* pColIndex, int32_t numOfCols) {
-  if (tsdbRLockRepoMeta(tsdb) < 0) goto _error;
-
-  STable* pTable = tsdbGetTableByUid(tsdbGetMeta(tsdb), uid);
-  if (pTable == NULL) {
-    tsdbError("%p failed to get stable, uid:%" PRIu64, tsdb, uid);
+                                 SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId) {
+  STbCfg* pTbCfg = metaGetTbInfoByUid(tsdb->pMeta, uid);
+  if (pTbCfg == NULL) {
+    tsdbError("%p failed to get stable, uid:%"PRIu64", reqId:0x%"PRIx64, tsdb, uid, reqId);
     terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-    tsdbUnlockRepoMeta(tsdb);
-
     goto _error;
   }
 
-  if (pTable->type != TSDB_SUPER_TABLE) {
-    tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", tid:%d, name:%s", tsdb, uid, pTable->tableId,
-        pTable->name->data);
-    terrno = TSDB_CODE_COM_OPS_NOT_SUPPORT; //basically, this error is caused by invalid sql issued by client
-
-    tsdbUnlockRepoMeta(tsdb);
+  if (pTbCfg->type != META_SUPER_TABLE) {
+    tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", reId:0x%"PRIx64, tsdb, uid, reqId);
+    terrno = TSDB_CODE_OPS_NOT_SUPPORT; //basically, this error is caused by invalid sql issued by client
     goto _error;
   }
 
   //NOTE: not add ref count for super table
   SArray* res = taosArrayInit(8, sizeof(STableKeyInfo));
-  STSchema* pTagSchema = tsdbGetTableTagSchema(pTable);
+  STSchema* pTagSchema = metaGetTableSchema(tsdb->pMeta, uid, 0, true);
 
   // no tags and tbname condition, all child tables of this stable are involved
   if (tbnameCond == NULL && (pTagCond == NULL || len == 0)) {
-    int32_t ret = getAllTableList(pTable, res);
+    assert(false);
+    int32_t ret = 0;//getAllTableList(pTable, res);
     if (ret != TSDB_CODE_SUCCESS) {
-      tsdbUnlockRepoMeta(tsdb);
       goto _error;
     }
 
@@ -3736,60 +3727,60 @@ int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const ch
               pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList));
 
     taosArrayDestroy(res);
-    if (tsdbUnlockRepoMeta(tsdb) < 0) goto _error;
     return ret;
   }
 
   int32_t ret = TSDB_CODE_SUCCESS;
-  tExprNode* expr = NULL;
-
-  TRY(TSDB_MAX_TAG_CONDITIONS) {
-    expr = exprTreeFromTableName(tbnameCond);
-    if (expr == NULL) {
-      expr = exprTreeFromBinary(pTagCond, len);
-    } else {
-      CLEANUP_PUSH_VOID_PTR_PTR(true, tExprTreeDestroy, expr, NULL);
-      tExprNode* tagExpr = exprTreeFromBinary(pTagCond, len);
-      if (tagExpr != NULL) {
-        CLEANUP_PUSH_VOID_PTR_PTR(true, tExprTreeDestroy, tagExpr, NULL);
-        tExprNode* tbnameExpr = expr;
-        expr = calloc(1, sizeof(tExprNode));
-        if (expr == NULL) {
-          THROW( TSDB_CODE_TDB_OUT_OF_MEMORY );
-        }
-        expr->nodeType = TSQL_NODE_EXPR;
-        expr->_node.optr = (uint8_t)tagNameRelType;
-        expr->_node.pLeft = tagExpr;
-        expr->_node.pRight = tbnameExpr;
-      }
-    }
-    CLEANUP_EXECUTE();
-
-  } CATCH( code ) {
-    CLEANUP_EXECUTE();
-    terrno = code;
-    tsdbUnlockRepoMeta(tsdb);     // unlock tsdb in any cases
-
-    goto _error;
-    // TODO: more error handling
-  } END_TRY
-
-  doQueryTableList(pTable, res, expr);
-  pGroupInfo->numOfTables = (uint32_t)taosArrayGetSize(res);
-  pGroupInfo->pGroupList  = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey);
-
-  tsdbDebug("%p stable tid:%d, uid:%"PRIu64" query, numOfTables:%u, belong to %" PRIzu " groups", tsdb, pTable->tableId,
-      pTable->uid, pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList));
-
-  taosArrayDestroy(res);
-
-  if (tsdbUnlockRepoMeta(tsdb) < 0) goto _error;
-  return ret;
+//  tExprNode* expr = NULL;
+//
+//  TRY(TSDB_MAX_TAG_CONDITIONS) {
+//    expr = exprTreeFromTableName(tbnameCond);
+//    if (expr == NULL) {
+//      expr = exprTreeFromBinary(pTagCond, len);
+//    } else {
+//      CLEANUP_PUSH_VOID_PTR_PTR(true, tExprTreeDestroy, expr, NULL);
+//      tExprNode* tagExpr = exprTreeFromBinary(pTagCond, len);
+//      if (tagExpr != NULL) {
+//        CLEANUP_PUSH_VOID_PTR_PTR(true, tExprTreeDestroy, tagExpr, NULL);
+//        tExprNode* tbnameExpr = expr;
+//        expr = calloc(1, sizeof(tExprNode));
+//        if (expr == NULL) {
+//          THROW( TSDB_CODE_TDB_OUT_OF_MEMORY );
+//        }
+//        expr->nodeType = TSQL_NODE_EXPR;
+//        expr->_node.optr = (uint8_t)tagNameRelType;
+//        expr->_node.pLeft = tagExpr;
+//        expr->_node.pRight = tbnameExpr;
+//      }
+//    }
+//    CLEANUP_EXECUTE();
+//
+//  } CATCH( code ) {
+//    CLEANUP_EXECUTE();
+//    terrno = code;
+//    tsdbUnlockRepoMeta(tsdb);     // unlock tsdb in any cases
+//
+//    goto _error;
+//    // TODO: more error handling
+//  } END_TRY
+//
+//  doQueryTableList(pTable, res, expr);
+//  pGroupInfo->numOfTables = (uint32_t)taosArrayGetSize(res);
+//  pGroupInfo->pGroupList  = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey);
+//
+//  tsdbDebug("%p stable tid:%d, uid:%"PRIu64" query, numOfTables:%u, belong to %" PRIzu " groups", tsdb, pTable->tableId,
+//      pTable->uid, pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList));
+//
+//  taosArrayDestroy(res);
+//
+//  if (tsdbUnlockRepoMeta(tsdb) < 0) goto _error;
+//  return ret;
 
   _error:
   return terrno;
 }
 
+#if 0
 int32_t tsdbGetOneTableGroup(STsdb* tsdb, uint64_t uid, TSKEY startKey, STableGroupInfo* pGroupInfo) {
   if (tsdbRLockRepoMeta(tsdb) < 0) goto _error;
 
diff --git a/source/libs/executor/inc/dataSinkInt.h b/source/libs/executor/inc/dataSinkInt.h
index 69727626af..7003564365 100644
--- a/source/libs/executor/inc/dataSinkInt.h
+++ b/source/libs/executor/inc/dataSinkInt.h
@@ -32,7 +32,7 @@ typedef struct SDataSinkManager {
 } SDataSinkManager;
 
 typedef int32_t (*FPutDataBlock)(struct SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue);
-typedef void (*FEndPut)(struct SDataSinkHandle* pHandle, int64_t useconds);
+typedef void (*FEndPut)(struct SDataSinkHandle* pHandle, uint64_t useconds);
 typedef void (*FGetDataLength)(struct SDataSinkHandle* pHandle, int32_t* pLen, bool* pQueryEnd);
 typedef int32_t (*FGetDataBlock)(struct SDataSinkHandle* pHandle, SOutputData* pOutput);
 typedef int32_t (*FDestroyDataSinker)(struct SDataSinkHandle* pHandle);
diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index 34d89a75b3..adb305ab09 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -597,7 +597,6 @@ void clearOutputBuf(SOptrBasicInfo* pBInfo, int32_t *bufCapacity);
 void copyTsColoum(SSDataBlock* pRes, SQLFunctionCtx* pCtx, int32_t numOfOutput);
 
 void freeParam(STaskParam *param);
-int32_t convertQueryMsg(SQueryTableReq *pQueryMsg, STaskParam* param);
 int32_t createQueryFunc(SQueriedTableInfo* pTableInfo, int32_t numOfOutput, SExprInfo** pExprInfo,
                         SSqlExpr** pExprMsg, SColumnInfo* pTagCols, int32_t queryType, void* pMsg, struct SUdfInfo* pUdfInfo);
 
@@ -638,7 +637,8 @@ size_t getResultSize(SQInfo *pQInfo, int64_t *numOfRows);
 void setQueryKilled(SQInfo *pQInfo);
 
 void publishOperatorProfEvent(SOperatorInfo* operatorInfo, EQueryProfEventType eventType);
-void publishQueryAbortEvent(SQInfo* pQInfo, int32_t code);
+void publishQueryAbortEvent(SExecTaskInfo * pTaskInfo, int32_t code);
+
 void calculateOperatorProfResults(SQInfo* pQInfo);
 void queryCostStatis(SQInfo *pQInfo);
 
diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c
index 8280f9d0af..edba4fc97d 100644
--- a/source/libs/executor/src/dataDispatcher.c
+++ b/source/libs/executor/src/dataDispatcher.c
@@ -44,7 +44,7 @@ typedef struct SDataDispatchHandle {
   SDataDispatchBuf nextOutput;
   int32_t status;
   bool queryEnd;
-  int64_t useconds;
+  uint64_t useconds;
   pthread_mutex_t mutex;
 } SDataDispatchHandle;
 
@@ -158,7 +158,7 @@ static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput,
   return TSDB_CODE_SUCCESS;
 }
 
-static void endPut(struct SDataSinkHandle* pHandle, int64_t useconds) {
+static void endPut(struct SDataSinkHandle* pHandle, uint64_t useconds) {
   SDataDispatchHandle* pDispatcher = (SDataDispatchHandle*)pHandle;
   pthread_mutex_lock(&pDispatcher->mutex);
   pDispatcher->queryEnd = true;
diff --git a/source/libs/executor/src/dataSinkMgt.c b/source/libs/executor/src/dataSinkMgt.c
index 80d99f96c6..eb1f75f359 100644
--- a/source/libs/executor/src/dataSinkMgt.c
+++ b/source/libs/executor/src/dataSinkMgt.c
@@ -37,7 +37,7 @@ int32_t dsPutDataBlock(DataSinkHandle handle, const SInputData* pInput, bool* pC
   return pHandleImpl->fPut(pHandleImpl, pInput, pContinue);
 }
 
-void dsEndPut(DataSinkHandle handle, int64_t useconds) {
+void dsEndPut(DataSinkHandle handle, uint64_t useconds) {
   SDataSinkHandle* pHandleImpl = (SDataSinkHandle*)handle;
   return pHandleImpl->fEndPut(pHandleImpl, useconds);
 }
diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c
index daeefba253..1f5d0cd059 100644
--- a/source/libs/executor/src/executorMain.c
+++ b/source/libs/executor/src/executorMain.c
@@ -68,7 +68,7 @@ void freeParam(STaskParam *param) {
   tfree(param->prevResult);
 }
 
-int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo) {
+int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_t* pTaskInfo, DataSinkHandle* handle) {
   assert(tsdb != NULL && pSubplan != NULL);
   SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;
 
@@ -85,6 +85,8 @@ int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_
 
   code = dsCreateDataSinker(pSubplan->pDataSink, &(*pTask)->dsHandle);
 
+  *handle = (*pTask)->dsHandle;
+
   _error:
   // if failed to add ref for all tables in this query, abort current query
   return code;
@@ -135,10 +137,12 @@ int waitMoment(SQInfo* pQInfo){
 }
 #endif
 
-int32_t qExecTask(qTaskInfo_t tinfo, DataSinkHandle* handle) {
+int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
   int64_t        threadId = taosGetSelfPthreadId();
 
+  *pRes = NULL;
+
   int64_t curOwner = 0;
   if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) {
     qError("QInfo:0x%" PRIx64 "-%p qhandle is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo,
@@ -153,7 +157,7 @@ int32_t qExecTask(qTaskInfo_t tinfo, DataSinkHandle* handle) {
 
   if (isTaskKilled(pTaskInfo)) {
     qDebug("QInfo:0x%" PRIx64 " it is already killed, abort", GET_TASKID(pTaskInfo));
-    return pTaskInfo->code;
+    return TSDB_CODE_SUCCESS;
   }
 
   //  STaskRuntimeEnv* pRuntimeEnv = &pTaskInfo->runtimeEnv;
@@ -168,7 +172,8 @@ int32_t qExecTask(qTaskInfo_t tinfo, DataSinkHandle* handle) {
   if (ret != TSDB_CODE_SUCCESS) {
     publishQueryAbortEvent(pTaskInfo, ret);
     pTaskInfo->code = ret;
-    qDebug("QInfo:0x%" PRIx64 " query abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));
+    qDebug("QInfo:0x%" PRIx64 " query abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo),
+           tstrerror(pTaskInfo->code));
     return pTaskInfo->code;
   }
 
@@ -178,39 +183,21 @@ int32_t qExecTask(qTaskInfo_t tinfo, DataSinkHandle* handle) {
   publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_BEFORE_OPERATOR_EXEC);
   int64_t st = 0;
 
-  if (handle) {
-    *handle = pTaskInfo->dsHandle;
+  st = taosGetTimestampUs();
+  *pRes = pTaskInfo->pRoot->exec(pTaskInfo->pRoot, &newgroup);
+
+  pTaskInfo->cost.elapsedTime += (taosGetTimestampUs() - st);
+  publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_AFTER_OPERATOR_EXEC);
+
+  if (NULL == *pRes) {
+    *useconds = pTaskInfo->cost.elapsedTime;
   }
-  
-  while(1) {
-    st = taosGetTimestampUs();
-    SSDataBlock* pRes = pTaskInfo->pRoot->exec(pTaskInfo->pRoot, &newgroup);
 
-    pTaskInfo->cost.elapsedTime += (taosGetTimestampUs() - st);
-    publishOperatorProfEvent(pTaskInfo->pRoot, QUERY_PROF_AFTER_OPERATOR_EXEC);
+  qDebug("QInfo:0x%" PRIx64 " query paused, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d",
+         GET_TASKID(pTaskInfo), 0, 0L, 0);
 
-    if (pRes == NULL) { // no results generated yet, abort
-      dsEndPut(pTaskInfo->dsHandle, pTaskInfo->cost.elapsedTime);
-      return pTaskInfo->code;
-    }
-
-    bool  qcontinue = false;
-    SInputData inputData = {.pData = pRes, .pTableRetrieveTsMap = NULL};
-    pTaskInfo->code = dsPutDataBlock(pTaskInfo->dsHandle, &inputData, &qcontinue);
-
-    if (isTaskKilled(pTaskInfo)) {
-      qDebug("QInfo:0x%" PRIx64 " task is killed", GET_TASKID(pTaskInfo));
-      //  } else if (GET_NUM_OF_RESULTS(pRuntimeEnv) == 0) {
-      //    qDebug("QInfo:0x%"PRIx64" over, %u tables queried, total %"PRId64" rows returned", pTaskInfo->qId, pRuntimeEnv->tableqinfoGroupInfo.numOfTables,
-      //           pRuntimeEnv->resultInfo.total);
-    }
-
-    if (!qcontinue) {
-          qDebug("QInfo:0x%"PRIx64" query paused, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d", GET_TASKID(pTaskInfo),
-              0, 0L, 0);
-      return pTaskInfo->code;
-    }
-  }
+  atomic_store_64(&pTaskInfo->owner, 0);
+  return pTaskInfo->code;
 }
 
 int32_t qRetrieveQueryResultInfo(qTaskInfo_t qinfo, bool* buildRes, void* pRspContext) {
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index a515804234..b6df0c527e 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -4186,14 +4186,14 @@ void publishOperatorProfEvent(SOperatorInfo* operatorInfo, EQueryProfEventType e
   }
 }
 
-void publishQueryAbortEvent(SQInfo* pQInfo, int32_t code) {
+void publishQueryAbortEvent(SExecTaskInfo * pTaskInfo, int32_t code) {
   SQueryProfEvent event;
   event.eventType = QUERY_PROF_QUERY_ABORT;
   event.eventTime = taosGetTimestampUs();
   event.abortCode = code;
 
-  if (pQInfo->summary.queryProfEvents) {
-    taosArrayPush(pQInfo->summary.queryProfEvents, &event);
+  if (pTaskInfo->cost.queryProfEvents) {
+    taosArrayPush(pTaskInfo->cost.queryProfEvents, &event);
   }
 }
 
@@ -7423,358 +7423,358 @@ int32_t doCreateExecTaskInfo(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, void* r
  * @param pExpr
  * @return
  */
-int32_t convertQueryMsg(SQueryTableReq *pQueryMsg, STaskParam* param) {
-  int32_t code = TSDB_CODE_SUCCESS;
-
-//  if (taosCheckVersion(pQueryMsg->version, version, 3) != 0) {
-//    return TSDB_CODE_QRY_INVALID_MSG;
+//int32_t convertQueryMsg(SQueryTableReq *pQueryMsg, STaskParam* param) {
+//  int32_t code = TSDB_CODE_SUCCESS;
+//
+////  if (taosCheckVersion(pQueryMsg->version, version, 3) != 0) {
+////    return TSDB_CODE_QRY_INVALID_MSG;
+////  }
+//
+//  pQueryMsg->numOfTables = htonl(pQueryMsg->numOfTables);
+//  pQueryMsg->window.skey = htobe64(pQueryMsg->window.skey);
+//  pQueryMsg->window.ekey = htobe64(pQueryMsg->window.ekey);
+//  pQueryMsg->interval.interval = htobe64(pQueryMsg->interval.interval);
+//  pQueryMsg->interval.sliding = htobe64(pQueryMsg->interval.sliding);
+//  pQueryMsg->interval.offset = htobe64(pQueryMsg->interval.offset);
+//  pQueryMsg->limit = htobe64(pQueryMsg->limit);
+//  pQueryMsg->offset = htobe64(pQueryMsg->offset);
+//  pQueryMsg->vgroupLimit = htobe64(pQueryMsg->vgroupLimit);
+//
+//  pQueryMsg->order = htons(pQueryMsg->order);
+//  pQueryMsg->orderColId = htons(pQueryMsg->orderColId);
+//  pQueryMsg->queryType = htonl(pQueryMsg->queryType);
+////  pQueryMsg->tagNameRelType = htons(pQueryMsg->tagNameRelType);
+//
+//  pQueryMsg->numOfCols = htons(pQueryMsg->numOfCols);
+//  pQueryMsg->numOfOutput = htons(pQueryMsg->numOfOutput);
+//  pQueryMsg->numOfGroupCols = htons(pQueryMsg->numOfGroupCols);
+//
+//  pQueryMsg->tagCondLen = htons(pQueryMsg->tagCondLen);
+//  pQueryMsg->colCondLen = htons(pQueryMsg->colCondLen);
+//
+//  pQueryMsg->tsBuf.tsOffset = htonl(pQueryMsg->tsBuf.tsOffset);
+//  pQueryMsg->tsBuf.tsLen = htonl(pQueryMsg->tsBuf.tsLen);
+//  pQueryMsg->tsBuf.tsNumOfBlocks = htonl(pQueryMsg->tsBuf.tsNumOfBlocks);
+//  pQueryMsg->tsBuf.tsOrder = htonl(pQueryMsg->tsBuf.tsOrder);
+//
+//  pQueryMsg->numOfTags = htonl(pQueryMsg->numOfTags);
+////  pQueryMsg->tbnameCondLen = htonl(pQueryMsg->tbnameCondLen);
+//  pQueryMsg->secondStageOutput = htonl(pQueryMsg->secondStageOutput);
+//  pQueryMsg->sqlstrLen = htonl(pQueryMsg->sqlstrLen);
+//  pQueryMsg->prevResultLen = htonl(pQueryMsg->prevResultLen);
+////  pQueryMsg->sw.gap = htobe64(pQueryMsg->sw.gap);
+////  pQueryMsg->sw.primaryColId = htonl(pQueryMsg->sw.primaryColId);
+//  pQueryMsg->tableScanOperator = htonl(pQueryMsg->tableScanOperator);
+//  pQueryMsg->numOfOperator = htonl(pQueryMsg->numOfOperator);
+//  pQueryMsg->udfContentOffset = htonl(pQueryMsg->udfContentOffset);
+//  pQueryMsg->udfContentLen    = htonl(pQueryMsg->udfContentLen);
+//  pQueryMsg->udfNum           = htonl(pQueryMsg->udfNum);
+//
+//  // query msg safety check
+//  if (!validateQueryMsg(pQueryMsg)) {
+//    code = TSDB_CODE_QRY_INVALID_MSG;
+//    goto _cleanup;
 //  }
-
-  pQueryMsg->numOfTables = htonl(pQueryMsg->numOfTables);
-  pQueryMsg->window.skey = htobe64(pQueryMsg->window.skey);
-  pQueryMsg->window.ekey = htobe64(pQueryMsg->window.ekey);
-  pQueryMsg->interval.interval = htobe64(pQueryMsg->interval.interval);
-  pQueryMsg->interval.sliding = htobe64(pQueryMsg->interval.sliding);
-  pQueryMsg->interval.offset = htobe64(pQueryMsg->interval.offset);
-  pQueryMsg->limit = htobe64(pQueryMsg->limit);
-  pQueryMsg->offset = htobe64(pQueryMsg->offset);
-  pQueryMsg->vgroupLimit = htobe64(pQueryMsg->vgroupLimit);
-
-  pQueryMsg->order = htons(pQueryMsg->order);
-  pQueryMsg->orderColId = htons(pQueryMsg->orderColId);
-  pQueryMsg->queryType = htonl(pQueryMsg->queryType);
-//  pQueryMsg->tagNameRelType = htons(pQueryMsg->tagNameRelType);
-
-  pQueryMsg->numOfCols = htons(pQueryMsg->numOfCols);
-  pQueryMsg->numOfOutput = htons(pQueryMsg->numOfOutput);
-  pQueryMsg->numOfGroupCols = htons(pQueryMsg->numOfGroupCols);
-
-  pQueryMsg->tagCondLen = htons(pQueryMsg->tagCondLen);
-  pQueryMsg->colCondLen = htons(pQueryMsg->colCondLen);  
-
-  pQueryMsg->tsBuf.tsOffset = htonl(pQueryMsg->tsBuf.tsOffset);
-  pQueryMsg->tsBuf.tsLen = htonl(pQueryMsg->tsBuf.tsLen);
-  pQueryMsg->tsBuf.tsNumOfBlocks = htonl(pQueryMsg->tsBuf.tsNumOfBlocks);
-  pQueryMsg->tsBuf.tsOrder = htonl(pQueryMsg->tsBuf.tsOrder);
-
-  pQueryMsg->numOfTags = htonl(pQueryMsg->numOfTags);
-//  pQueryMsg->tbnameCondLen = htonl(pQueryMsg->tbnameCondLen);
-  pQueryMsg->secondStageOutput = htonl(pQueryMsg->secondStageOutput);
-  pQueryMsg->sqlstrLen = htonl(pQueryMsg->sqlstrLen);
-  pQueryMsg->prevResultLen = htonl(pQueryMsg->prevResultLen);
-//  pQueryMsg->sw.gap = htobe64(pQueryMsg->sw.gap);
-//  pQueryMsg->sw.primaryColId = htonl(pQueryMsg->sw.primaryColId);
-  pQueryMsg->tableScanOperator = htonl(pQueryMsg->tableScanOperator);
-  pQueryMsg->numOfOperator = htonl(pQueryMsg->numOfOperator);
-  pQueryMsg->udfContentOffset = htonl(pQueryMsg->udfContentOffset);
-  pQueryMsg->udfContentLen    = htonl(pQueryMsg->udfContentLen);
-  pQueryMsg->udfNum           = htonl(pQueryMsg->udfNum);
-
-  // query msg safety check
-  if (!validateQueryMsg(pQueryMsg)) {
-    code = TSDB_CODE_QRY_INVALID_MSG;
-    goto _cleanup;
-  }
-
-  char *pMsg = (char *)(pQueryMsg->tableCols) + sizeof(SColumnInfo) * pQueryMsg->numOfCols;
-  for (int32_t col = 0; col < pQueryMsg->numOfCols; ++col) {
-    SColumnInfo *pColInfo = &pQueryMsg->tableCols[col];
-
-    pColInfo->colId = htons(pColInfo->colId);
-    pColInfo->type = htons(pColInfo->type);
-    pColInfo->bytes = htons(pColInfo->bytes);
-    pColInfo->flist.numOfFilters = 0;
-
-    if (!isValidDataType(pColInfo->type)) {
-      //qDebug("qmsg:%p, invalid data type in source column, index:%d, type:%d", pQueryMsg, col, pColInfo->type);
-      code = TSDB_CODE_QRY_INVALID_MSG;
-      goto _cleanup;
-    }
-
-/*
-    int32_t numOfFilters = pColInfo->flist.numOfFilters;
-    if (numOfFilters > 0) {
-      pColInfo->flist.filterInfo = calloc(numOfFilters, sizeof(SColumnFilterInfo));
-      if (pColInfo->flist.filterInfo == NULL) {
-        code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-        goto _cleanup;
-      }
-    }
-
-    code = deserializeColFilterInfo(pColInfo->flist.filterInfo, numOfFilters, &pMsg);
-    if (code != TSDB_CODE_SUCCESS) {
-      goto _cleanup;
-    }
-*/    
-  }
-
-  if (pQueryMsg->colCondLen > 0) {
-    param->colCond = calloc(1, pQueryMsg->colCondLen);
-    if (param->colCond == NULL) {
-      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-      goto _cleanup;
-    }
-
-    memcpy(param->colCond, pMsg, pQueryMsg->colCondLen);
-    pMsg += pQueryMsg->colCondLen;
-  }
-
-
-  param->tableScanOperator = pQueryMsg->tableScanOperator;
-  param->pExpr = calloc(pQueryMsg->numOfOutput, POINTER_BYTES);
-  if (param->pExpr == NULL) {
-    code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-    goto _cleanup;
-  }
-
-  SSqlExpr *pExprMsg = (SSqlExpr *)pMsg;
-
-  for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
-    param->pExpr[i] = pExprMsg;
-
-//    pExprMsg->colInfo.colIndex = htons(pExprMsg->colInfo.colIndex);
-//    pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
-//    pExprMsg->colInfo.flag  = htons(pExprMsg->colInfo.flag);
-//    pExprMsg->colBytes      = htons(pExprMsg->colBytes);
-//    pExprMsg->colType       = htons(pExprMsg->colType);
-
-//    pExprMsg->resType       = htons(pExprMsg->resType);
-//    pExprMsg->resBytes      = htons(pExprMsg->resBytes);
-    pExprMsg->interBytes    = htonl(pExprMsg->interBytes);
-
-//    pExprMsg->functionId    = htons(pExprMsg->functionId);
-    pExprMsg->numOfParams   = htons(pExprMsg->numOfParams);
-//    pExprMsg->resColId      = htons(pExprMsg->resColId);
-//    pExprMsg->flist.numOfFilters  = htons(pExprMsg->flist.numOfFilters);
-    pMsg += sizeof(SSqlExpr);
-
-    for (int32_t j = 0; j < pExprMsg->numOfParams; ++j) {
-      pExprMsg->param[j].nType = htonl(pExprMsg->param[j].nType);
-      pExprMsg->param[j].nLen  = htonl(pExprMsg->param[j].nLen);
-
-      if (pExprMsg->param[j].nType == TSDB_DATA_TYPE_BINARY) {
-        pExprMsg->param[j].pz = pMsg;
-        pMsg += pExprMsg->param[j].nLen;  // one more for the string terminated char.
-      } else {
-        pExprMsg->param[j].i = htobe64(pExprMsg->param[j].i);
-      }
-    }
-
-//    int16_t functionId = pExprMsg->functionId;
-//    if (functionId == FUNCTION_TAG || functionId == FUNCTION_TAGPRJ || functionId == FUNCTION_TAG_DUMMY) {
-//      if (!TSDB_COL_IS_TAG(pExprMsg->colInfo.flag)) {  // ignore the column  index check for arithmetic expression.
-//        code = TSDB_CODE_QRY_INVALID_MSG;
+//
+//  char *pMsg = (char *)(pQueryMsg->tableCols) + sizeof(SColumnInfo) * pQueryMsg->numOfCols;
+//  for (int32_t col = 0; col < pQueryMsg->numOfCols; ++col) {
+//    SColumnInfo *pColInfo = &pQueryMsg->tableCols[col];
+//
+//    pColInfo->colId = htons(pColInfo->colId);
+//    pColInfo->type = htons(pColInfo->type);
+//    pColInfo->bytes = htons(pColInfo->bytes);
+//    pColInfo->flist.numOfFilters = 0;
+//
+//    if (!isValidDataType(pColInfo->type)) {
+//      //qDebug("qmsg:%p, invalid data type in source column, index:%d, type:%d", pQueryMsg, col, pColInfo->type);
+//      code = TSDB_CODE_QRY_INVALID_MSG;
+//      goto _cleanup;
+//    }
+//
+///*
+//    int32_t numOfFilters = pColInfo->flist.numOfFilters;
+//    if (numOfFilters > 0) {
+//      pColInfo->flist.filterInfo = calloc(numOfFilters, sizeof(SColumnFilterInfo));
+//      if (pColInfo->flist.filterInfo == NULL) {
+//        code = TSDB_CODE_QRY_OUT_OF_MEMORY;
 //        goto _cleanup;
 //      }
 //    }
-
-//    if (pExprMsg->flist.numOfFilters > 0) {
-//      pExprMsg->flist.filterInfo = calloc(pExprMsg->flist.numOfFilters, sizeof(SColumnFilterInfo));
-//    }
 //
-//    deserializeColFilterInfo(pExprMsg->flist.filterInfo, pExprMsg->flist.numOfFilters, &pMsg);
-    pExprMsg = (SSqlExpr *)pMsg;
-  }
-
-  if (pQueryMsg->secondStageOutput) {
-    pExprMsg = (SSqlExpr *)pMsg;
-    param->pSecExpr = calloc(pQueryMsg->secondStageOutput, POINTER_BYTES);
-
-    for (int32_t i = 0; i < pQueryMsg->secondStageOutput; ++i) {
-      param->pSecExpr[i] = pExprMsg;
-
-//      pExprMsg->colInfo.colIndex = htons(pExprMsg->colInfo.colIndex);
-//      pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
-//      pExprMsg->colInfo.flag  = htons(pExprMsg->colInfo.flag);
-//      pExprMsg->resType       = htons(pExprMsg->resType);
-//      pExprMsg->resBytes      = htons(pExprMsg->resBytes);
-//      pExprMsg->colBytes      = htons(pExprMsg->colBytes);
-//      pExprMsg->colType       = htons(pExprMsg->colType);
-
-//      pExprMsg->functionId = htons(pExprMsg->functionId);
-      pExprMsg->numOfParams = htons(pExprMsg->numOfParams);
-
-      pMsg += sizeof(SSqlExpr);
-
-      for (int32_t j = 0; j < pExprMsg->numOfParams; ++j) {
-        pExprMsg->param[j].nType = htonl(pExprMsg->param[j].nType);
-        pExprMsg->param[j].nLen = htonl(pExprMsg->param[j].nLen);
-
-        if (pExprMsg->param[j].nType == TSDB_DATA_TYPE_BINARY) {
-          pExprMsg->param[j].pz = pMsg;
-          pMsg += pExprMsg->param[j].nLen;  // one more for the string terminated char.
-        } else {
-          pExprMsg->param[j].i = htobe64(pExprMsg->param[j].i);
-        }
-      }
-
-//      int16_t functionId = pExprMsg->functionId;
-//      if (functionId == FUNCTION_TAG || functionId == FUNCTION_TAGPRJ || functionId == FUNCTION_TAG_DUMMY) {
-//        if (!TSDB_COL_IS_TAG(pExprMsg->colInfo.flag)) {  // ignore the column  index check for arithmetic expression.
-//          code = TSDB_CODE_QRY_INVALID_MSG;
-//          goto _cleanup;
-//        }
-//      }
-
-      pExprMsg = (SSqlExpr *)pMsg;
-    }
-  }
-
-  pMsg = createTableIdList(pQueryMsg, pMsg, &(param->pTableIdList));
-
-  if (pQueryMsg->numOfGroupCols > 0) {  // group by tag columns
-    param->pGroupColIndex = malloc(pQueryMsg->numOfGroupCols * sizeof(SColIndex));
-    if (param->pGroupColIndex == NULL) {
-      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-      goto _cleanup;
-    }
-
-    for (int32_t i = 0; i < pQueryMsg->numOfGroupCols; ++i) {
-      param->pGroupColIndex[i].colId = htons(*(int16_t *)pMsg);
-      pMsg += sizeof(param->pGroupColIndex[i].colId);
-
-      param->pGroupColIndex[i].colIndex = htons(*(int16_t *)pMsg);
-      pMsg += sizeof(param->pGroupColIndex[i].colIndex);
-
-      param->pGroupColIndex[i].flag = htons(*(int16_t *)pMsg);
-      pMsg += sizeof(param->pGroupColIndex[i].flag);
-
-      memcpy(param->pGroupColIndex[i].name, pMsg, tListLen(param->pGroupColIndex[i].name));
-      pMsg += tListLen(param->pGroupColIndex[i].name);
-    }
-
-    pQueryMsg->orderByIdx = htons(pQueryMsg->orderByIdx);
-    pQueryMsg->orderType = htons(pQueryMsg->orderType);
-  }
-
-  pQueryMsg->fillType = htons(pQueryMsg->fillType);
-  if (pQueryMsg->fillType != TSDB_FILL_NONE) {
-    pQueryMsg->fillVal = (uint64_t)(pMsg);
-
-    int64_t *v = (int64_t *)pMsg;
-    for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
-      v[i] = htobe64(v[i]);
-    }
-
-    pMsg += sizeof(int64_t) * pQueryMsg->numOfOutput;
-  }
-
-  if (pQueryMsg->numOfTags > 0) {
-    param->pTagColumnInfo = calloc(1, sizeof(SColumnInfo) * pQueryMsg->numOfTags);
-    if (param->pTagColumnInfo == NULL) {
-      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-      goto _cleanup;
-    }
-
-    for (int32_t i = 0; i < pQueryMsg->numOfTags; ++i) {
-      SColumnInfo* pTagCol = (SColumnInfo*) pMsg;
-
-      pTagCol->colId = htons(pTagCol->colId);
-      pTagCol->bytes = htons(pTagCol->bytes);
-      pTagCol->type  = htons(pTagCol->type);
-//      pTagCol->flist.numOfFilters = 0;
-
-      param->pTagColumnInfo[i] = *pTagCol;
-      pMsg += sizeof(SColumnInfo);
-    }
-  }
-
-  // the tag query condition expression string is located at the end of query msg
-  if (pQueryMsg->tagCondLen > 0) {
-    param->tagCond = calloc(1, pQueryMsg->tagCondLen);
-    if (param->tagCond == NULL) {
-      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-      goto _cleanup;
-    }
-
-    memcpy(param->tagCond, pMsg, pQueryMsg->tagCondLen);
-    pMsg += pQueryMsg->tagCondLen;
-  }
-
-  if (pQueryMsg->prevResultLen > 0) {
-    param->prevResult = calloc(1, pQueryMsg->prevResultLen);
-    if (param->prevResult == NULL) {
-      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-      goto _cleanup;
-    }
-
-    memcpy(param->prevResult, pMsg, pQueryMsg->prevResultLen);
-    pMsg += pQueryMsg->prevResultLen;
-  }
-
-//  if (pQueryMsg->tbnameCondLen > 0) {
-//    param->tbnameCond = calloc(1, pQueryMsg->tbnameCondLen + 1);
-//    if (param->tbnameCond == NULL) {
+//    code = deserializeColFilterInfo(pColInfo->flist.filterInfo, numOfFilters, &pMsg);
+//    if (code != TSDB_CODE_SUCCESS) {
+//      goto _cleanup;
+//    }
+//*/
+//  }
+//
+//  if (pQueryMsg->colCondLen > 0) {
+//    param->colCond = calloc(1, pQueryMsg->colCondLen);
+//    if (param->colCond == NULL) {
 //      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
 //      goto _cleanup;
 //    }
 //
-//    strncpy(param->tbnameCond, pMsg, pQueryMsg->tbnameCondLen);
-//    pMsg += pQueryMsg->tbnameCondLen;
+//    memcpy(param->colCond, pMsg, pQueryMsg->colCondLen);
+//    pMsg += pQueryMsg->colCondLen;
 //  }
-
-  //skip ts buf
-  if ((pQueryMsg->tsBuf.tsOffset + pQueryMsg->tsBuf.tsLen) > 0) {
-    pMsg = (char *)pQueryMsg + pQueryMsg->tsBuf.tsOffset + pQueryMsg->tsBuf.tsLen;
-  }
-
-  param->pOperator = taosArrayInit(pQueryMsg->numOfOperator, sizeof(int32_t));
-  for(int32_t i = 0; i < pQueryMsg->numOfOperator; ++i) {
-    int32_t op = htonl(*(int32_t*)pMsg);
-    taosArrayPush(param->pOperator, &op);
-
-    pMsg += sizeof(int32_t);
-  }
-
-  if (pQueryMsg->udfContentLen > 0) {
-    // todo extract udf function in tudf.c
-//    param->pUdfInfo = calloc(1, sizeof(SUdfInfo));
-//    param->pUdfInfo->contLen = pQueryMsg->udfContentLen;
 //
-//    pMsg = (char*)pQueryMsg + pQueryMsg->udfContentOffset;
-//    param->pUdfInfo->resType = *(int8_t*) pMsg;
-//    pMsg += sizeof(int8_t);
 //
-//    param->pUdfInfo->resBytes = htons(*(int16_t*)pMsg);
-//    pMsg += sizeof(int16_t);
+//  param->tableScanOperator = pQueryMsg->tableScanOperator;
+//  param->pExpr = calloc(pQueryMsg->numOfOutput, POINTER_BYTES);
+//  if (param->pExpr == NULL) {
+//    code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+//    goto _cleanup;
+//  }
 //
-//    tstr* name = (tstr*)(pMsg);
-//    param->pUdfInfo->name = strndup(name->data, name->len);
+//  SSqlExpr *pExprMsg = (SSqlExpr *)pMsg;
+//
+//  for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
+//    param->pExpr[i] = pExprMsg;
+//
+////    pExprMsg->colInfo.colIndex = htons(pExprMsg->colInfo.colIndex);
+////    pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
+////    pExprMsg->colInfo.flag  = htons(pExprMsg->colInfo.flag);
+////    pExprMsg->colBytes      = htons(pExprMsg->colBytes);
+////    pExprMsg->colType       = htons(pExprMsg->colType);
+//
+////    pExprMsg->resType       = htons(pExprMsg->resType);
+////    pExprMsg->resBytes      = htons(pExprMsg->resBytes);
+//    pExprMsg->interBytes    = htonl(pExprMsg->interBytes);
+//
+////    pExprMsg->functionId    = htons(pExprMsg->functionId);
+//    pExprMsg->numOfParams   = htons(pExprMsg->numOfParams);
+////    pExprMsg->resColId      = htons(pExprMsg->resColId);
+////    pExprMsg->flist.numOfFilters  = htons(pExprMsg->flist.numOfFilters);
+//    pMsg += sizeof(SSqlExpr);
+//
+//    for (int32_t j = 0; j < pExprMsg->numOfParams; ++j) {
+//      pExprMsg->param[j].nType = htonl(pExprMsg->param[j].nType);
+//      pExprMsg->param[j].nLen  = htonl(pExprMsg->param[j].nLen);
+//
+//      if (pExprMsg->param[j].nType == TSDB_DATA_TYPE_BINARY) {
+//        pExprMsg->param[j].pz = pMsg;
+//        pMsg += pExprMsg->param[j].nLen;  // one more for the string terminated char.
+//      } else {
+//        pExprMsg->param[j].i = htobe64(pExprMsg->param[j].i);
+//      }
+//    }
+//
+////    int16_t functionId = pExprMsg->functionId;
+////    if (functionId == FUNCTION_TAG || functionId == FUNCTION_TAGPRJ || functionId == FUNCTION_TAG_DUMMY) {
+////      if (!TSDB_COL_IS_TAG(pExprMsg->colInfo.flag)) {  // ignore the column  index check for arithmetic expression.
+////        code = TSDB_CODE_QRY_INVALID_MSG;
+////        goto _cleanup;
+////      }
+////    }
+//
+////    if (pExprMsg->flist.numOfFilters > 0) {
+////      pExprMsg->flist.filterInfo = calloc(pExprMsg->flist.numOfFilters, sizeof(SColumnFilterInfo));
+////    }
+////
+////    deserializeColFilterInfo(pExprMsg->flist.filterInfo, pExprMsg->flist.numOfFilters, &pMsg);
+//    pExprMsg = (SSqlExpr *)pMsg;
+//  }
+//
+//  if (pQueryMsg->secondStageOutput) {
+//    pExprMsg = (SSqlExpr *)pMsg;
+//    param->pSecExpr = calloc(pQueryMsg->secondStageOutput, POINTER_BYTES);
+//
+//    for (int32_t i = 0; i < pQueryMsg->secondStageOutput; ++i) {
+//      param->pSecExpr[i] = pExprMsg;
+//
+////      pExprMsg->colInfo.colIndex = htons(pExprMsg->colInfo.colIndex);
+////      pExprMsg->colInfo.colId = htons(pExprMsg->colInfo.colId);
+////      pExprMsg->colInfo.flag  = htons(pExprMsg->colInfo.flag);
+////      pExprMsg->resType       = htons(pExprMsg->resType);
+////      pExprMsg->resBytes      = htons(pExprMsg->resBytes);
+////      pExprMsg->colBytes      = htons(pExprMsg->colBytes);
+////      pExprMsg->colType       = htons(pExprMsg->colType);
+//
+////      pExprMsg->functionId = htons(pExprMsg->functionId);
+//      pExprMsg->numOfParams = htons(pExprMsg->numOfParams);
+//
+//      pMsg += sizeof(SSqlExpr);
+//
+//      for (int32_t j = 0; j < pExprMsg->numOfParams; ++j) {
+//        pExprMsg->param[j].nType = htonl(pExprMsg->param[j].nType);
+//        pExprMsg->param[j].nLen = htonl(pExprMsg->param[j].nLen);
+//
+//        if (pExprMsg->param[j].nType == TSDB_DATA_TYPE_BINARY) {
+//          pExprMsg->param[j].pz = pMsg;
+//          pMsg += pExprMsg->param[j].nLen;  // one more for the string terminated char.
+//        } else {
+//          pExprMsg->param[j].i = htobe64(pExprMsg->param[j].i);
+//        }
+//      }
+//
+////      int16_t functionId = pExprMsg->functionId;
+////      if (functionId == FUNCTION_TAG || functionId == FUNCTION_TAGPRJ || functionId == FUNCTION_TAG_DUMMY) {
+////        if (!TSDB_COL_IS_TAG(pExprMsg->colInfo.flag)) {  // ignore the column  index check for arithmetic expression.
+////          code = TSDB_CODE_QRY_INVALID_MSG;
+////          goto _cleanup;
+////        }
+////      }
+//
+//      pExprMsg = (SSqlExpr *)pMsg;
+//    }
+//  }
+//
+//  pMsg = createTableIdList(pQueryMsg, pMsg, &(param->pTableIdList));
+//
+//  if (pQueryMsg->numOfGroupCols > 0) {  // group by tag columns
+//    param->pGroupColIndex = malloc(pQueryMsg->numOfGroupCols * sizeof(SColIndex));
+//    if (param->pGroupColIndex == NULL) {
+//      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+//      goto _cleanup;
+//    }
+//
+//    for (int32_t i = 0; i < pQueryMsg->numOfGroupCols; ++i) {
+//      param->pGroupColIndex[i].colId = htons(*(int16_t *)pMsg);
+//      pMsg += sizeof(param->pGroupColIndex[i].colId);
+//
+//      param->pGroupColIndex[i].colIndex = htons(*(int16_t *)pMsg);
+//      pMsg += sizeof(param->pGroupColIndex[i].colIndex);
+//
+//      param->pGroupColIndex[i].flag = htons(*(int16_t *)pMsg);
+//      pMsg += sizeof(param->pGroupColIndex[i].flag);
+//
+//      memcpy(param->pGroupColIndex[i].name, pMsg, tListLen(param->pGroupColIndex[i].name));
+//      pMsg += tListLen(param->pGroupColIndex[i].name);
+//    }
+//
+//    pQueryMsg->orderByIdx = htons(pQueryMsg->orderByIdx);
+//    pQueryMsg->orderType = htons(pQueryMsg->orderType);
+//  }
+//
+//  pQueryMsg->fillType = htons(pQueryMsg->fillType);
+//  if (pQueryMsg->fillType != TSDB_FILL_NONE) {
+//    pQueryMsg->fillVal = (uint64_t)(pMsg);
+//
+//    int64_t *v = (int64_t *)pMsg;
+//    for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
+//      v[i] = htobe64(v[i]);
+//    }
+//
+//    pMsg += sizeof(int64_t) * pQueryMsg->numOfOutput;
+//  }
+//
+//  if (pQueryMsg->numOfTags > 0) {
+//    param->pTagColumnInfo = calloc(1, sizeof(SColumnInfo) * pQueryMsg->numOfTags);
+//    if (param->pTagColumnInfo == NULL) {
+//      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+//      goto _cleanup;
+//    }
+//
+//    for (int32_t i = 0; i < pQueryMsg->numOfTags; ++i) {
+//      SColumnInfo* pTagCol = (SColumnInfo*) pMsg;
+//
+//      pTagCol->colId = htons(pTagCol->colId);
+//      pTagCol->bytes = htons(pTagCol->bytes);
+//      pTagCol->type  = htons(pTagCol->type);
+////      pTagCol->flist.numOfFilters = 0;
+//
+//      param->pTagColumnInfo[i] = *pTagCol;
+//      pMsg += sizeof(SColumnInfo);
+//    }
+//  }
+//
+//  // the tag query condition expression string is located at the end of query msg
+//  if (pQueryMsg->tagCondLen > 0) {
+//    param->tagCond = calloc(1, pQueryMsg->tagCondLen);
+//    if (param->tagCond == NULL) {
+//      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+//      goto _cleanup;
+//    }
+//
+//    memcpy(param->tagCond, pMsg, pQueryMsg->tagCondLen);
+//    pMsg += pQueryMsg->tagCondLen;
+//  }
+//
+//  if (pQueryMsg->prevResultLen > 0) {
+//    param->prevResult = calloc(1, pQueryMsg->prevResultLen);
+//    if (param->prevResult == NULL) {
+//      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+//      goto _cleanup;
+//    }
+//
+//    memcpy(param->prevResult, pMsg, pQueryMsg->prevResultLen);
+//    pMsg += pQueryMsg->prevResultLen;
+//  }
+//
+////  if (pQueryMsg->tbnameCondLen > 0) {
+////    param->tbnameCond = calloc(1, pQueryMsg->tbnameCondLen + 1);
+////    if (param->tbnameCond == NULL) {
+////      code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+////      goto _cleanup;
+////    }
+////
+////    strncpy(param->tbnameCond, pMsg, pQueryMsg->tbnameCondLen);
+////    pMsg += pQueryMsg->tbnameCondLen;
+////  }
+//
+//  //skip ts buf
+//  if ((pQueryMsg->tsBuf.tsOffset + pQueryMsg->tsBuf.tsLen) > 0) {
+//    pMsg = (char *)pQueryMsg + pQueryMsg->tsBuf.tsOffset + pQueryMsg->tsBuf.tsLen;
+//  }
+//
+//  param->pOperator = taosArrayInit(pQueryMsg->numOfOperator, sizeof(int32_t));
+//  for(int32_t i = 0; i < pQueryMsg->numOfOperator; ++i) {
+//    int32_t op = htonl(*(int32_t*)pMsg);
+//    taosArrayPush(param->pOperator, &op);
 //
-//    pMsg += varDataTLen(name);
-//    param->pUdfInfo->funcType = htonl(*(int32_t*)pMsg);
 //    pMsg += sizeof(int32_t);
+//  }
 //
-//    param->pUdfInfo->bufSize = htonl(*(int32_t*)pMsg);
-//    pMsg += sizeof(int32_t);
+//  if (pQueryMsg->udfContentLen > 0) {
+//    // todo extract udf function in tudf.c
+////    param->pUdfInfo = calloc(1, sizeof(SUdfInfo));
+////    param->pUdfInfo->contLen = pQueryMsg->udfContentLen;
+////
+////    pMsg = (char*)pQueryMsg + pQueryMsg->udfContentOffset;
+////    param->pUdfInfo->resType = *(int8_t*) pMsg;
+////    pMsg += sizeof(int8_t);
+////
+////    param->pUdfInfo->resBytes = htons(*(int16_t*)pMsg);
+////    pMsg += sizeof(int16_t);
+////
+////    tstr* name = (tstr*)(pMsg);
+////    param->pUdfInfo->name = strndup(name->data, name->len);
+////
+////    pMsg += varDataTLen(name);
+////    param->pUdfInfo->funcType = htonl(*(int32_t*)pMsg);
+////    pMsg += sizeof(int32_t);
+////
+////    param->pUdfInfo->bufSize = htonl(*(int32_t*)pMsg);
+////    pMsg += sizeof(int32_t);
+////
+////    param->pUdfInfo->content = malloc(pQueryMsg->udfContentLen);
+////    memcpy(param->pUdfInfo->content, pMsg, pQueryMsg->udfContentLen);
 //
-//    param->pUdfInfo->content = malloc(pQueryMsg->udfContentLen);
-//    memcpy(param->pUdfInfo->content, pMsg, pQueryMsg->udfContentLen);
-
-    pMsg += pQueryMsg->udfContentLen;
-  }
-
-  param->sql = strndup(pMsg, pQueryMsg->sqlstrLen);
-
-  SQueriedTableInfo info = { .numOfTags = pQueryMsg->numOfTags, .numOfCols = pQueryMsg->numOfCols, .colList = pQueryMsg->tableCols};
-  if (!validateQueryTableCols(&info, param->pExpr, pQueryMsg->numOfOutput, param->pTagColumnInfo, pQueryMsg)) {
-    code = TSDB_CODE_QRY_INVALID_MSG;
-    goto _cleanup;
-  }
-
-  //qDebug("qmsg:%p query %d tables, type:%d, qrange:%" PRId64 "-%" PRId64 ", numOfGroupbyTagCols:%d, order:%d, "
-//         "outputCols:%d, numOfCols:%d, interval:%" PRId64 ", fillType:%d, comptsLen:%d, compNumOfBlocks:%d, limit:%" PRId64 ", offset:%" PRId64,
-//         pQueryMsg, pQueryMsg->numOfTables, pQueryMsg->queryType, pQueryMsg->window.skey, pQueryMsg->window.ekey, pQueryMsg->numOfGroupCols,
-//         pQueryMsg->order, pQueryMsg->numOfOutput, pQueryMsg->numOfCols, pQueryMsg->interval.interval,
-//         pQueryMsg->fillType, pQueryMsg->tsBuf.tsLen, pQueryMsg->tsBuf.tsNumOfBlocks, pQueryMsg->limit, pQueryMsg->offset);
-
-  //qDebug("qmsg:%p, sql:%s", pQueryMsg, param->sql);
-  return TSDB_CODE_SUCCESS;
-
-_cleanup:
-  freeParam(param);
-  return code;
-}
+//    pMsg += pQueryMsg->udfContentLen;
+//  }
+//
+//  param->sql = strndup(pMsg, pQueryMsg->sqlstrLen);
+//
+//  SQueriedTableInfo info = { .numOfTags = pQueryMsg->numOfTags, .numOfCols = pQueryMsg->numOfCols, .colList = pQueryMsg->tableCols};
+//  if (!validateQueryTableCols(&info, param->pExpr, pQueryMsg->numOfOutput, param->pTagColumnInfo, pQueryMsg)) {
+//    code = TSDB_CODE_QRY_INVALID_MSG;
+//    goto _cleanup;
+//  }
+//
+//  //qDebug("qmsg:%p query %d tables, type:%d, qrange:%" PRId64 "-%" PRId64 ", numOfGroupbyTagCols:%d, order:%d, "
+////         "outputCols:%d, numOfCols:%d, interval:%" PRId64 ", fillType:%d, comptsLen:%d, compNumOfBlocks:%d, limit:%" PRId64 ", offset:%" PRId64,
+////         pQueryMsg, pQueryMsg->numOfTables, pQueryMsg->queryType, pQueryMsg->window.skey, pQueryMsg->window.ekey, pQueryMsg->numOfGroupCols,
+////         pQueryMsg->order, pQueryMsg->numOfOutput, pQueryMsg->numOfCols, pQueryMsg->interval.interval,
+////         pQueryMsg->fillType, pQueryMsg->tsBuf.tsLen, pQueryMsg->tsBuf.tsNumOfBlocks, pQueryMsg->limit, pQueryMsg->offset);
+//
+//  //qDebug("qmsg:%p, sql:%s", pQueryMsg, param->sql);
+//  return TSDB_CODE_SUCCESS;
+//
+//_cleanup:
+//  freeParam(param);
+//  return code;
+//}
 
 int32_t cloneExprFilterInfo(SColumnFilterInfo **dst, SColumnFilterInfo* src, int32_t filterNum) {
   if (filterNum <= 0) {
diff --git a/source/libs/executor/test/executorTests.cpp b/source/libs/executor/test/executorTests.cpp
index 5f5fddbe28..c528d879a3 100644
--- a/source/libs/executor/test/executorTests.cpp
+++ b/source/libs/executor/test/executorTests.cpp
@@ -217,5 +217,6 @@ TEST(testCase, build_executor_tree_Test) {
       "}";
 
   SExecTaskInfo* pTaskInfo = nullptr;
-  int32_t code = qCreateExecTask((void*) 1, 2, NULL, (void**) &pTaskInfo);
+  DataSinkHandle sinkHandle = nullptr;
+  int32_t code = qCreateExecTask((void*) 1, 2, NULL, (void**) &pTaskInfo, &sinkHandle);
 }
\ No newline at end of file
diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c
index 09e9af9788..a0beaba61d 100644
--- a/source/libs/qworker/src/qworker.c
+++ b/source/libs/qworker/src/qworker.c
@@ -458,6 +458,37 @@ _return:
   QW_RET(code);
 }
 
+int32_t qwExecTask(QW_FPARAMS_DEF, qTaskInfo_t taskHandle, DataSinkHandle sinkHandle) {
+  int32_t code = 0;
+  bool  qcontinue = true;
+  SSDataBlock* pRes = NULL;
+  uint64_t useconds = 0;
+  
+  while (qcontinue) {
+    code = qExecTask(taskHandle, &pRes, &useconds);
+    if (code) {
+      QW_TASK_ELOG("qExecTask failed, code:%x", code);
+      QW_ERR_JRET(code);
+    }
+
+    if (NULL == pRes) {
+      QW_TASK_DLOG("query done, useconds:%"PRIu64, useconds);
+      dsEndPut(sinkHandle, useconds);
+      break;
+    }
+
+    SInputData inputData = {.pData = pRes, .pTableRetrieveTsMap = NULL};
+    code = dsPutDataBlock(sinkHandle, &inputData, &qcontinue);
+    if (code) {
+      QW_TASK_ELOG("dsPutDataBlock failed, code:%x", code);
+      QW_ERR_JRET(code);
+    }
+  }
+
+_return:
+
+  QW_RET(code);
+}
 
 
 int32_t qwGetResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, void **rspMsg, SOutputData *pOutput) {
@@ -733,7 +764,9 @@ int32_t qwProcessQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
   }
 
   qTaskInfo_t pTaskInfo = NULL;
-  code = qCreateExecTask(qwMsg->node, 0, (struct SSubplan *)plan, &pTaskInfo);
+  DataSinkHandle sinkHandle = NULL;
+  
+  code = qCreateExecTask(qwMsg->node, 0, (struct SSubplan *)plan, &pTaskInfo, &sinkHandle);
   if (code) {
     QW_TASK_ELOG("qCreateExecTask failed, code:%x", code);
     QW_ERR_JRET(code);
@@ -743,12 +776,7 @@ int32_t qwProcessQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
 
   queryRsped = true;
 
-  DataSinkHandle sinkHandle = NULL;
-  code = qExecTask(pTaskInfo, &sinkHandle);
-  if (code) {
-    QW_TASK_ELOG("qExecTask failed, code:%x", code);
-    QW_ERR_JRET(code);
-  }
+  QW_ERR_JRET(qwExecTask(QW_FPARAMS(), pTaskInfo, sinkHandle));
 
 _return:
 
@@ -840,11 +868,7 @@ int32_t qwProcessCQuery(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
   qTaskInfo_t     taskHandle = ctx->taskHandle;
   DataSinkHandle  sinkHandle = ctx->sinkHandle;
 
-  code = qExecTask(taskHandle, &sinkHandle);
-  if (code) {
-    QW_TASK_ELOG("qExecTask failed, code:%x", code);
-    QW_ERR_JRET(code);
-  }
+  QW_ERR_JRET(qwExecTask(QW_FPARAMS(), taskHandle, sinkHandle));
 
   QW_SET_EVENT_PROCESSED(ctx, QW_EVENT_CQUERY);
 
diff --git a/source/libs/scheduler/src/scheduler.c b/source/libs/scheduler/src/scheduler.c
index 2d5322fc2c..c53926f8c1 100644
--- a/source/libs/scheduler/src/scheduler.c
+++ b/source/libs/scheduler/src/scheduler.c
@@ -412,6 +412,8 @@ int32_t schSetTaskCandidateAddrs(SSchJob *pJob, SSchTask *pTask) {
         SCH_TASK_ELOG("taosArrayPush execNode to candidate addrs failed, addNum:%d, errno:%d", addNum, errno);
         SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
+
+      ++addNum;
     }
   }
 
@@ -792,6 +794,11 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
         if (rspCode != TSDB_CODE_SUCCESS) {
           SCH_ERR_RET(schProcessOnTaskFailure(pJob, pTask, rspCode));
         }
+
+        SShellSubmitRsp *rsp = (SShellSubmitRsp *)msg;
+        if (rsp) {
+          pJob->resNumOfRows += rsp->affectedRows;
+        }
         #endif
 
         SCH_ERR_RET(schProcessOnTaskSuccess(pJob, pTask));
@@ -1355,9 +1362,9 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, stru
 
   SSchJob *job = NULL;
 
-  SCH_ERR_RET(schExecJobImpl(transport, nodeList, pDag, &job, true));
+  SCH_ERR_RET(schExecJobImpl(transport, nodeList, pDag, pJob, true));
 
-  *pJob = job;
+  job = *pJob;
 
   pRes->code = atomic_load_32(&job->errCode);
   pRes->numOfRows = job->resNumOfRows;
diff --git a/source/libs/scheduler/test/schedulerTests.cpp b/source/libs/scheduler/test/schedulerTests.cpp
index d72c4920d5..5332c6fcd1 100644
--- a/source/libs/scheduler/test/schedulerTests.cpp
+++ b/source/libs/scheduler/test/schedulerTests.cpp
@@ -34,10 +34,12 @@
 #include "stub.h"
 #include "addr_any.h"
 
+
 namespace {
 
 extern "C" int32_t schHandleResponseMsg(SSchJob *job, SSchTask *task, int32_t msgType, char *msg, int32_t msgSize, int32_t rspCode);
 
+
 void schtInitLogFile() {
   const char    *defaultLogFileNamePrefix = "taoslog";
   const int32_t  maxLogFileNum = 10;
@@ -113,9 +115,9 @@ void schtBuildInsertDag(SQueryDag *dag) {
   dag->queryId = qId;
   dag->numOfSubplans = 2;
   dag->pSubplans = taosArrayInit(1, POINTER_BYTES);
-  SArray *inserta = taosArrayInit(dag->numOfSubplans, sizeof(SSubplan));
+  SArray *inserta = taosArrayInit(dag->numOfSubplans, POINTER_BYTES);
   
-  SSubplan insertPlan[2] = {0};
+  SSubplan *insertPlan = (SSubplan *)calloc(2, sizeof(SSubplan));
 
   insertPlan[0].id.queryId = qId;
   insertPlan[0].id.templateId = 0x0000000000000003;
@@ -131,6 +133,7 @@ void schtBuildInsertDag(SQueryDag *dag) {
   insertPlan[0].pParents = NULL;
   insertPlan[0].pNode = NULL;
   insertPlan[0].pDataSink = (SDataSink*)calloc(1, sizeof(SDataSink));
+  insertPlan[0].msgType = TDMT_VND_SUBMIT;
 
   insertPlan[1].id.queryId = qId;
   insertPlan[1].id.templateId = 0x0000000000000003;
@@ -146,10 +149,11 @@ void schtBuildInsertDag(SQueryDag *dag) {
   insertPlan[1].pParents = NULL;
   insertPlan[1].pNode = NULL;
   insertPlan[1].pDataSink = (SDataSink*)calloc(1, sizeof(SDataSink));
+  insertPlan[1].msgType = TDMT_VND_SUBMIT;
 
-
-  taosArrayPush(inserta, &insertPlan[0]);
-  taosArrayPush(inserta, &insertPlan[1]);
+  taosArrayPush(inserta, &insertPlan);
+  insertPlan += 1;
+  taosArrayPush(inserta, &insertPlan);
 
   taosArrayPush(dag->pSubplans, &inserta);  
 }
@@ -210,6 +214,24 @@ void schtSetRpcSendRequest() {
   }
 }
 
+int32_t schtAsyncSendMsgToServer(void *pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo) {
+  return 0;
+}
+
+
+void schtSetAsyncSendMsgToServer() {
+  static Stub stub;
+  stub.set(asyncSendMsgToServer, schtAsyncSendMsgToServer);
+  {
+    AddrAny any("libtransport.so");
+    std::map<std::string,void*> result;
+    any.get_global_func_addr_dynsym("^asyncSendMsgToServer$", result);
+    for (const auto& f : result) {
+      stub.set(f.second, schtAsyncSendMsgToServer);
+    }
+  }
+}
+
 
 void *schtSendRsp(void *param) {
   SSchJob *job = NULL;
@@ -230,7 +252,7 @@ void *schtSendRsp(void *param) {
 
     SShellSubmitRsp rsp = {0};
     rsp.affectedRows = 10;
-    schHandleResponseMsg(job, task, TDMT_VND_SUBMIT, (char *)&rsp, sizeof(rsp), 0);
+    schHandleResponseMsg(job, task, TDMT_VND_SUBMIT_RSP, (char *)&rsp, sizeof(rsp), 0);
     
     pIter = taosHashIterate(job->execTasks, pIter);
   }    
@@ -238,6 +260,23 @@ void *schtSendRsp(void *param) {
   return NULL;
 }
 
+void *schtCreateFetchRspThread(void *param) {
+  struct SSchJob* job = (struct SSchJob*)param;
+
+  sleep(1);
+
+  int32_t code = 0;
+  SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)calloc(1, sizeof(SRetrieveTableRsp));
+  rsp->completed = 1;
+  rsp->numOfRows = 10;
+  code = schHandleResponseMsg(job, job->fetchTask, TDMT_VND_FETCH_RSP, (char *)rsp, sizeof(rsp), 0);
+    
+  assert(code == 0);
+}
+
+
+
+
 struct SSchJob *pInsertJob = NULL;
 }
 
@@ -266,6 +305,7 @@ TEST(queryTest, normalCase) {
 
   schtSetPlanToString();
   schtSetExecNode();
+  schtSetAsyncSendMsgToServer();
   
   code = scheduleAsyncExecJob(mockPointer, qnodeList, &dag, &pJob);
   ASSERT_EQ(code, 0);
@@ -276,7 +316,7 @@ TEST(queryTest, normalCase) {
     SSchTask *task = *(SSchTask **)pIter;
 
     SQueryTableRsp rsp = {0};
-    code = schHandleResponseMsg(job, task, TDMT_VND_QUERY, (char *)&rsp, sizeof(rsp), 0);
+    code = schHandleResponseMsg(job, task, TDMT_VND_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0);
     
     ASSERT_EQ(code, 0);
     pIter = taosHashIterate(job->execTasks, pIter);
@@ -287,8 +327,8 @@ TEST(queryTest, normalCase) {
     SSchTask *task = *(SSchTask **)pIter;
 
     SResReadyRsp rsp = {0};
-    code = schHandleResponseMsg(job, task, TDMT_VND_RES_READY, (char *)&rsp, sizeof(rsp), 0);
-    
+    code = schHandleResponseMsg(job, task, TDMT_VND_RES_READY_RSP, (char *)&rsp, sizeof(rsp), 0);
+    printf("code:%d", code);
     ASSERT_EQ(code, 0);
     pIter = taosHashIterate(job->execTasks, pIter);
   }  
@@ -298,7 +338,7 @@ TEST(queryTest, normalCase) {
     SSchTask *task = *(SSchTask **)pIter;
 
     SQueryTableRsp rsp = {0};
-    code = schHandleResponseMsg(job, task, TDMT_VND_QUERY, (char *)&rsp, sizeof(rsp), 0);
+    code = schHandleResponseMsg(job, task, TDMT_VND_QUERY_RSP, (char *)&rsp, sizeof(rsp), 0);
     
     ASSERT_EQ(code, 0);
     pIter = taosHashIterate(job->execTasks, pIter);
@@ -309,22 +349,19 @@ TEST(queryTest, normalCase) {
     SSchTask *task = *(SSchTask **)pIter;
 
     SResReadyRsp rsp = {0};
-    code = schHandleResponseMsg(job, task, TDMT_VND_RES_READY, (char *)&rsp, sizeof(rsp), 0);
+    code = schHandleResponseMsg(job, task, TDMT_VND_RES_READY_RSP, (char *)&rsp, sizeof(rsp), 0);
     ASSERT_EQ(code, 0);
     
     pIter = taosHashIterate(job->execTasks, pIter);
   }  
 
-  SRetrieveTableRsp rsp = {0};
-  rsp.completed = 1;
-  rsp.numOfRows = 10;
-  code = schHandleResponseMsg(job, NULL, TDMT_VND_FETCH, (char *)&rsp, sizeof(rsp), 0);
-    
-  ASSERT_EQ(code, 0);
+  pthread_attr_t thattr;
+  pthread_attr_init(&thattr);
 
+  pthread_t thread1;
+  pthread_create(&(thread1), &thattr, schtCreateFetchRspThread, job);
 
-  void *data = NULL;
-  
+  void *data = NULL;  
   code = scheduleFetchRows(job, &data);
   ASSERT_EQ(code, 0);
 
@@ -340,6 +377,8 @@ TEST(queryTest, normalCase) {
   scheduleFreeJob(pJob);
 
   schtFreeQueryDag(&dag);
+
+  schedulerDestroy();
 }
 
 
@@ -369,6 +408,7 @@ TEST(insertTest, normalCase) {
   schtBuildInsertDag(&dag);
 
   schtSetPlanToString();
+  schtSetAsyncSendMsgToServer();
 
   pthread_attr_t thattr;
   pthread_attr_init(&thattr);
@@ -382,6 +422,8 @@ TEST(insertTest, normalCase) {
   ASSERT_EQ(res.numOfRows, 20);
 
   scheduleFreeJob(pInsertJob);
+
+  schedulerDestroy();  
 }
 
 TEST(multiThread, forceFree) {