From ad6a648d2b3ecb53e5b1debccf890ed6277e8d7f Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sat, 4 Jun 2022 17:08:46 +0800
Subject: [PATCH 1/8] refactor: do some internal refactor.

---
 source/client/inc/clientInt.h      | 3 +--
 source/libs/catalog/src/ctgAsync.c | 6 +++---
 2 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h
index c7b831893e..09116a6bd1 100644
--- a/source/client/inc/clientInt.h
+++ b/source/client/inc/clientInt.h
@@ -45,8 +45,7 @@ extern "C" {
 
 #define ERROR_MSG_BUF_DEFAULT_SIZE 512
 #define HEARTBEAT_INTERVAL         1500  // ms
-
-#define SYNC_ON_TOP_OF_ASYNC 0
+#define SYNC_ON_TOP_OF_ASYNC       0
 
 enum {
   RES_TYPE__QUERY = 1,
diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c
index a9e808e749..ce94eace66 100644
--- a/source/libs/catalog/src/ctgAsync.c
+++ b/source/libs/catalog/src/ctgAsync.c
@@ -44,7 +44,7 @@ int32_t ctgInitGetTbMetaTask(SCtgJob *pJob, int32_t taskIdx, SName *name) {
 
   taosArrayPush(pJob->pTasks, &task);
 
-  qDebug("QID:%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, task.type, name->tname);
+  qDebug("QID:0x%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, task.type, name->tname);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -143,7 +143,7 @@ int32_t ctgInitGetTbHashTask(SCtgJob *pJob, int32_t taskIdx, SName *name) {
 
   taosArrayPush(pJob->pTasks, &task);
 
-  qDebug("QID:%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, task.type, name->tname);
+  qDebug("QID:0x%" PRIx64 " task %d type %d initialized, tableName:%s", pJob->queryId, taskIdx, task.type, name->tname);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -247,7 +247,7 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
 
   *taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dbCfgNum + indexNum + userNum + dbInfoNum;
   if (*taskNum <= 0) {
-    ctgError("Empty input for job, no need to retrieve meta, reqId:0x%" PRIx64, reqId);
+    ctgDebug("Empty input for job, no need to retrieve meta, reqId:0x%" PRIx64, reqId);
     return TSDB_CODE_SUCCESS;
   }
 

From d261c17a4de8c9a787106ffb17dfe481f3ed5919 Mon Sep 17 00:00:00 2001
From: Xiaoyu Wang <xiaoyuwang@taosdata.com>
Date: Sat, 4 Jun 2022 21:31:07 +0800
Subject: [PATCH 2/8] fix: a problem of parser async

---
 source/libs/parser/inc/parInt.h           |  1 -
 source/libs/parser/src/parAstParser.c     | 15 ++--
 source/libs/parser/src/parInsert.c        | 64 +++++++++-------
 source/libs/parser/src/parTranslater.c    | 91 ++++++++++++-----------
 source/libs/parser/src/parUtil.c          | 21 ++++--
 source/libs/parser/test/parInsertTest.cpp |  1 +
 6 files changed, 110 insertions(+), 83 deletions(-)

diff --git a/source/libs/parser/inc/parInt.h b/source/libs/parser/inc/parInt.h
index 8ec20cde5a..3efe6700d2 100644
--- a/source/libs/parser/inc/parInt.h
+++ b/source/libs/parser/inc/parInt.h
@@ -32,7 +32,6 @@ int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery);
 int32_t translate(SParseContext* pParseCxt, SQuery* pQuery);
 int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
 int32_t calculateConstant(SParseContext* pParseCxt, SQuery* pQuery);
-int32_t isNotSchemalessDb(SParseContext* pContext, char *dbName);
 
 #ifdef __cplusplus
 }
diff --git a/source/libs/parser/src/parAstParser.c b/source/libs/parser/src/parAstParser.c
index 7abe244261..96fb810e5c 100644
--- a/source/libs/parser/src/parAstParser.c
+++ b/source/libs/parser/src/parAstParser.c
@@ -175,11 +175,11 @@ static int32_t collectMetaKeyFromSelect(SCollectMetaKeyCxt* pCxt, SSelectStmt* p
 }
 
 static int32_t collectMetaKeyFromCreateTable(SCollectMetaKeyCxt* pCxt, SCreateTableStmt* pStmt) {
-  if (NULL == pStmt->pTags) {
-    return reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
-  } else {
-    return reserveDbCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pCxt->pMetaCache);
+  int32_t code = reserveDbCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pCxt->pMetaCache);
+  if (TSDB_CODE_SUCCESS == code && NULL == pStmt->pTags) {
+    code = reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
   }
+  return code;
 }
 
 static int32_t collectMetaKeyFromCreateMultiTable(SCollectMetaKeyCxt* pCxt, SCreateMultiTableStmt* pStmt) {
@@ -187,8 +187,11 @@ static int32_t collectMetaKeyFromCreateMultiTable(SCollectMetaKeyCxt* pCxt, SCre
   SNode*  pNode = NULL;
   FOREACH(pNode, pStmt->pSubTables) {
     SCreateSubTableClause* pClause = (SCreateSubTableClause*)pNode;
-    code =
-        reserveTableMetaInCache(pCxt->pParseCxt->acctId, pClause->useDbName, pClause->useTableName, pCxt->pMetaCache);
+    code = reserveDbCfgInCache(pCxt->pParseCxt->acctId, pClause->dbName, pCxt->pMetaCache);
+    if (TSDB_CODE_SUCCESS == code) {
+      code =
+          reserveTableMetaInCache(pCxt->pParseCxt->acctId, pClause->useDbName, pClause->useTableName, pCxt->pMetaCache);
+    }
     if (TSDB_CODE_SUCCESS == code) {
       code = reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
     }
diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c
index 422c480397..efe0bd75b2 100644
--- a/source/libs/parser/src/parInsert.c
+++ b/source/libs/parser/src/parInsert.c
@@ -254,7 +254,7 @@ static int32_t createSName(SName* pName, SToken* pTableName, int32_t acctId, con
 
 static int32_t checkAuth(SInsertParseContext* pCxt, char* pDbFname, bool* pPass) {
   SParseContext* pBasicCtx = pCxt->pComCxt;
-  if (NULL != pCxt->pMetaCache) {
+  if (pBasicCtx->async) {
     return getUserAuthFromCache(pCxt->pMetaCache, pBasicCtx->pUser, pDbFname, AUTH_TYPE_WRITE, pPass);
   }
   return catalogChkAuth(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pBasicCtx->pUser, pDbFname,
@@ -263,7 +263,7 @@ static int32_t checkAuth(SInsertParseContext* pCxt, char* pDbFname, bool* pPass)
 
 static int32_t getTableSchema(SInsertParseContext* pCxt, SName* pTbName, bool isStb, STableMeta** pTableMeta) {
   SParseContext* pBasicCtx = pCxt->pComCxt;
-  if (NULL != pCxt->pMetaCache) {
+  if (pBasicCtx->async) {
     return getTableMetaFromCache(pCxt->pMetaCache, pTbName, pTableMeta);
   }
   if (isStb) {
@@ -275,7 +275,7 @@ static int32_t getTableSchema(SInsertParseContext* pCxt, SName* pTbName, bool is
 
 static int32_t getTableVgroup(SInsertParseContext* pCxt, SName* pTbName, SVgroupInfo* pVg) {
   SParseContext* pBasicCtx = pCxt->pComCxt;
-  if (NULL != pCxt->pMetaCache) {
+  if (pBasicCtx->async) {
     return getTableVgroupFromCache(pCxt->pMetaCache, pTbName, pVg);
   }
   return catalogGetTableHashVgroup(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pTbName, pVg);
@@ -305,6 +305,16 @@ static int32_t getSTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFna
   return getTableMetaImpl(pCxt, name, dbFname, true);
 }
 
+static int32_t getDBCfg(SInsertParseContext* pCxt, const char* pDbFName, SDbCfgInfo* pInfo) {
+  SParseContext* pBasicCtx = pCxt->pComCxt;
+  if (pBasicCtx->async) {
+    CHECK_CODE(getDbCfgFromCache(pCxt->pMetaCache, pDbFName, pInfo));
+  } else {
+    CHECK_CODE(catalogGetDBCfg(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pDbFName, pInfo));
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pSchema) {
   while (start < end) {
     if (strlen(pSchema[start].name) == pColname->n && strncmp(pColname->z, pSchema[start].name, pColname->n) == 0) {
@@ -1090,10 +1100,10 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tb
 
   SName sname;
   createSName(&sname, &sToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg);
-  char stbFName[TSDB_TABLE_FNAME_LEN];
-  tNameExtractFullName(&sname, stbFName);
+  char dbFName[TSDB_DB_FNAME_LEN];
+  tNameGetFullDbName(&sname, dbFName);
 
-  CHECK_CODE(getSTableMeta(pCxt, &sname, stbFName));
+  CHECK_CODE(getSTableMeta(pCxt, &sname, dbFName));
   if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
     return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
   }
@@ -1282,6 +1292,14 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) {
   destroyBlockArrayList(pCxt->pVgDataBlocks);
 }
 
+static int32_t checkSchemalessDb(SInsertParseContext* pCxt, char* pDbName) {
+  SDbCfgInfo pInfo = {0};
+  char       fullName[TSDB_TABLE_FNAME_LEN];
+  snprintf(fullName, sizeof(fullName), "%d.%s", pCxt->pComCxt->acctId, pDbName);
+  CHECK_CODE(getDBCfg(pCxt, fullName, &pInfo));
+  return pInfo.schemaless ? TSDB_CODE_SML_INVALID_DB_CONF : TSDB_CODE_SUCCESS;
+}
+
 //   tb_name
 //       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
 //       [(field1_name, ...)]
@@ -1335,7 +1353,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
     SName name;
     CHECK_CODE(createSName(&name, &tbnameToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
 
-    CHECK_CODE(isNotSchemalessDb(pCxt->pComCxt, name.dbname));
+    CHECK_CODE(checkSchemalessDb(pCxt, name.dbname));
 
     tNameExtractFullName(&name, tbFName);
     CHECK_CODE(taosHashPut(pCxt->pTableNameHashObj, tbFName, strlen(tbFName), &name, sizeof(SName)));
@@ -1413,23 +1431,6 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
   return buildOutput(pCxt);
 }
 
-int32_t isNotSchemalessDb(SParseContext* pContext, char *dbName){
-  SName          name;
-  tNameSetDbName(&name, pContext->acctId, dbName, strlen(dbName));
-  char dbFname[TSDB_DB_FNAME_LEN] = {0};
-  tNameGetFullDbName(&name, dbFname);
-  SDbCfgInfo pInfo = {0};
-  int32_t code = catalogGetDBCfg(pContext->pCatalog, pContext->pTransporter, &pContext->mgmtEpSet, dbFname, &pInfo);
-  if (code != TSDB_CODE_SUCCESS) {
-    parserError("catalogGetDBCfg error, code:%s, dbFName:%s", tstrerror(code), dbFname);
-    return code;
-  }
-  if (pInfo.schemaless){
-    parserError("can not insert into schemaless db:%s", dbFname);
-    return TSDB_CODE_SML_INVALID_DB_CONF;
-  }
-  return TSDB_CODE_SUCCESS;
-}
 // INSERT INTO
 //   tb_name
 //       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
@@ -1472,6 +1473,8 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
     if (NULL == *pQuery) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
+  } else {
+    context.pMetaCache = (*pQuery)->pMetaCache;
   }
   (*pQuery)->execMode = QUERY_EXEC_MODE_SCHEDULE;
   (*pQuery)->haveResultSet = false;
@@ -1586,12 +1589,20 @@ static int32_t skipUsingClause(SInsertParseSyntaxCxt* pCxt) {
 static int32_t collectTableMetaKey(SInsertParseSyntaxCxt* pCxt, SToken* pTbToken) {
   SName name;
   CHECK_CODE(createSName(&name, pTbToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
+  CHECK_CODE(reserveDbCfgInCache(pCxt->pComCxt->acctId, name.dbname, pCxt->pMetaCache));
   CHECK_CODE(reserveUserAuthInCacheExt(pCxt->pComCxt->pUser, &name, AUTH_TYPE_WRITE, pCxt->pMetaCache));
   CHECK_CODE(reserveTableMetaInCacheExt(&name, pCxt->pMetaCache));
   CHECK_CODE(reserveTableVgroupInCacheExt(&name, pCxt->pMetaCache));
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t collectAutoCreateTableMetaKey(SInsertParseSyntaxCxt* pCxt, SToken* pTbToken) {
+  SName name;
+  CHECK_CODE(createSName(&name, pTbToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
+  CHECK_CODE(reserveTableVgroupInCacheExt(&name, pCxt->pMetaCache));
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t parseInsertBodySyntax(SInsertParseSyntaxCxt* pCxt) {
   bool hasData = false;
   // for each table
@@ -1620,6 +1631,7 @@ static int32_t parseInsertBodySyntax(SInsertParseSyntaxCxt* pCxt) {
 
     // USING clause
     if (TK_USING == sToken.type) {
+      CHECK_CODE(collectAutoCreateTableMetaKey(pCxt, &tbnameToken));
       NEXT_TOKEN(pCxt->pSql, sToken);
       CHECK_CODE(collectTableMetaKey(pCxt, &sToken));
       CHECK_CODE(skipUsingClause(pCxt));
@@ -2172,8 +2184,8 @@ static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* p
       val.nData = kv->length;
     } else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
       int32_t output = 0;
-      void *p = taosMemoryCalloc(1, kv->length * TSDB_NCHAR_SIZE);
-      if(p == NULL){
+      void*   p = taosMemoryCalloc(1, kv->length * TSDB_NCHAR_SIZE);
+      if (p == NULL) {
         code = TSDB_CODE_OUT_OF_MEMORY;
         goto end;
       }
diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index 2d37e9087d..69e45e2dcf 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -829,8 +829,9 @@ static EDealRes translateComparisonOperator(STranslateContext* pCxt, SOperatorNo
     if (!IS_VAR_DATA_TYPE(((SExprNode*)(pOp->pLeft))->resType.type)) {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
     }
-    if (QUERY_NODE_VALUE != nodeType(pOp->pRight) || 
-      ((!IS_STR_DATA_TYPE(((SExprNode*)(pOp->pRight))->resType.type)) && (((SExprNode*)(pOp->pRight))->resType.type != TSDB_DATA_TYPE_NULL))) {
+    if (QUERY_NODE_VALUE != nodeType(pOp->pRight) ||
+        ((!IS_STR_DATA_TYPE(((SExprNode*)(pOp->pRight))->resType.type)) &&
+         (((SExprNode*)(pOp->pRight))->resType.type != TSDB_DATA_TYPE_NULL))) {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName);
     }
   }
@@ -2632,8 +2633,23 @@ static int32_t checkTableSchema(STranslateContext* pCxt, SCreateTableStmt* pStmt
   return code;
 }
 
+static int32_t checkSchemalessDb(STranslateContext* pCxt, const char* pDbName) {
+  if (0 != pCxt->pParseCxt->schemalessType) {
+    return TSDB_CODE_SUCCESS;
+  }
+  SDbCfgInfo info = {0};
+  int32_t    code = getDBCfg(pCxt, pDbName, &info);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = info.schemaless ? TSDB_CODE_SML_INVALID_DB_CONF : TSDB_CODE_SUCCESS;
+  }
+  return code;
+}
+
 static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt) {
-  int32_t code = checTableFactorOption(pCxt, pStmt->pOptions->filesFactor);
+  int32_t code = checkSchemalessDb(pCxt, pStmt->dbName);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = checTableFactorOption(pCxt, pStmt->pOptions->filesFactor);
+  }
   if (TSDB_CODE_SUCCESS == code) {
     code = checkTableRollupOption(pCxt, pStmt->pOptions->pRollupFuncs);
   }
@@ -2646,11 +2662,6 @@ static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt
   if (TSDB_CODE_SUCCESS == code) {
     code = checkTableSchema(pCxt, pStmt);
   }
-  if (TSDB_CODE_SUCCESS == code) {
-    if(pCxt->pParseCxt->schemalessType == 0){
-      code = isNotSchemalessDb(pCxt->pParseCxt, pStmt->dbName);
-    }
-  }
   return code;
 }
 
@@ -4238,7 +4249,7 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
   int16_t  nTags = 0, nBufPos = 0;
   SSchema* pTagSchema = getTableTagSchema(pSuperTableMeta);
   SNode *  pTag = NULL, *pNode = NULL;
-  bool isJson = false;
+  bool     isJson = false;
   FORBOTH(pTag, pStmt->pSpecificTags, pNode, pStmt->pValsOfTags) {
     SColumnNode* pCol = (SColumnNode*)pTag;
     SSchema*     pSchema = NULL;
@@ -4271,11 +4282,11 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
 
       isJson = true;
       code = parseJsontoTagData(pVal->literal, pTagArray, ppTag, &pCxt->msgBuf);
-      if(code != TSDB_CODE_SUCCESS){
+      if (code != TSDB_CODE_SUCCESS) {
         goto end;
       }
-    }else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
-      void* nodeVal = nodesGetValueFromNode(pVal);
+    } else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
+      void*   nodeVal = nodesGetValueFromNode(pVal);
       STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
       if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
         val.pData = varDataVal(nodeVal);
@@ -4287,13 +4298,13 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla
     }
   }
 
-  if(!isJson) code = tTagNew(pTagArray, 1, false, ppTag);
+  if (!isJson) code = tTagNew(pTagArray, 1, false, ppTag);
 
 end:
-  if(isJson){
+  if (isJson) {
     for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
-      STagVal *p = (STagVal *)taosArrayGet(pTagArray, i);
-      if(IS_VAR_DATA_TYPE(p->type)){
+      STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
+      if (IS_VAR_DATA_TYPE(p->type)) {
         taosMemoryFree(p->pData);
       }
     }
@@ -4338,11 +4349,11 @@ static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClau
 
       isJson = true;
       code = parseJsontoTagData(pVal->literal, pTagArray, ppTag, &pCxt->msgBuf);
-      if(code != TSDB_CODE_SUCCESS){
+      if (code != TSDB_CODE_SUCCESS) {
         goto end;
       }
-    }else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
-      char* tmpVal = nodesGetValueFromNode(pVal);
+    } else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) {
+      char*   tmpVal = nodesGetValueFromNode(pVal);
       STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
       if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
         val.pData = varDataVal(tmpVal);
@@ -4354,13 +4365,13 @@ static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClau
     }
     ++index;
   }
-  if(!isJson) code = tTagNew(pTagArray, 1, false, ppTag);
+  if (!isJson) code = tTagNew(pTagArray, 1, false, ppTag);
 
 end:
-  if(isJson){
+  if (isJson) {
     for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
-      STagVal *p = (STagVal *)taosArrayGet(pTagArray, i);
-      if(IS_VAR_DATA_TYPE(p->type)){
+      STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
+      if (IS_VAR_DATA_TYPE(p->type)) {
         taosMemoryFree(p->pData);
       }
     }
@@ -4428,7 +4439,6 @@ static SArray* serializeVgroupsCreateTableBatch(int32_t acctId, SHashObj* pVgrou
 }
 
 static int32_t rewriteCreateMultiTable(STranslateContext* pCxt, SQuery* pQuery) {
-
   SCreateMultiTableStmt* pStmt = (SCreateMultiTableStmt*)pQuery->pRoot;
 
   SHashObj* pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
@@ -4439,11 +4449,11 @@ static int32_t rewriteCreateMultiTable(STranslateContext* pCxt, SQuery* pQuery)
   int32_t code = TSDB_CODE_SUCCESS;
   SNode*  pNode;
   FOREACH(pNode, pStmt->pSubTables) {
-    if(pCxt->pParseCxt->schemalessType == 0 &&
-        (code = isNotSchemalessDb(pCxt->pParseCxt, ((SCreateSubTableClause*)pNode)->dbName)) != TSDB_CODE_SUCCESS){
-      return code;
+    SCreateSubTableClause* pClause = (SCreateSubTableClause*)pNode;
+    code = checkSchemalessDb(pCxt, pClause->dbName);
+    if (TSDB_CODE_SUCCESS == code) {
+      code = rewriteCreateSubTable(pCxt, pClause, pVgroupHashmap);
     }
-    code = rewriteCreateSubTable(pCxt, (SCreateSubTableClause*)pNode, pVgroupHashmap);
     if (TSDB_CODE_SUCCESS != code) {
       taosHashCleanup(pVgroupHashmap);
       return code;
@@ -4633,27 +4643,27 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS
         strlen(pStmt->pVal->literal) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
       return buildSyntaxErrMsg(&pCxt->msgBuf, "json string too long than 4095", pStmt->pVal->literal);
     }
-    SArray *pTagVals = taosArrayInit(1, sizeof(STagVal));
+    SArray* pTagVals = taosArrayInit(1, sizeof(STagVal));
     int32_t code = TSDB_CODE_SUCCESS;
-    STag* pTag = NULL;
-    do{
+    STag*   pTag = NULL;
+    do {
       code = parseJsontoTagData(pStmt->pVal->literal, pTagVals, &pTag, &pCxt->msgBuf);
       if (TSDB_CODE_SUCCESS != code) {
         break;
       }
-    }while(0);
+    } while (0);
     for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
-      STagVal *p = (STagVal *)taosArrayGet(pTagVals, i);
-      if(IS_VAR_DATA_TYPE(p->type)){
+      STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
+      if (IS_VAR_DATA_TYPE(p->type)) {
         taosMemoryFree(p->pData);
       }
     }
     taosArrayDestroy(pTagVals);
-    if (code != TSDB_CODE_SUCCESS){
+    if (code != TSDB_CODE_SUCCESS) {
       return code;
     }
     pReq->nTagVal = pTag->len;
-    pReq->pTagVal = (uint8_t *)pTag;
+    pReq->pTagVal = (uint8_t*)pTag;
     pStmt->pVal->datum.p = (char*)pTag;  // for free
   } else {
     pReq->nTagVal = pStmt->pVal->node.resType.bytes;
@@ -4854,13 +4864,8 @@ static int32_t buildModifyVnodeArray(STranslateContext* pCxt, SAlterTableStmt* p
 
 static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
   SAlterTableStmt* pStmt = (SAlterTableStmt*)pQuery->pRoot;
-  int32_t     code = TSDB_CODE_SUCCESS;
-  if(pCxt->pParseCxt->schemalessType == 0 &&
-      (code = isNotSchemalessDb(pCxt->pParseCxt, pStmt->dbName)) != TSDB_CODE_SUCCESS){
-    return code;
-  }
-
-  STableMeta* pTableMeta = NULL;
+  int32_t          code = checkSchemalessDb(pCxt, pStmt->dbName);
+  STableMeta*      pTableMeta = NULL;
   code = getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pTableMeta);
   if (TSDB_CODE_SUCCESS != code) {
     return code;
diff --git a/source/libs/parser/src/parUtil.c b/source/libs/parser/src/parUtil.c
index 0a1915d6c2..4896b64686 100644
--- a/source/libs/parser/src/parUtil.c
+++ b/source/libs/parser/src/parUtil.c
@@ -322,11 +322,11 @@ static bool isValidateTag(char* input) {
   return true;
 }
 
-int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag **ppTag, SMsgBuf* pMsgBuf) {
+int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag** ppTag, SMsgBuf* pMsgBuf) {
   int32_t   retCode = TSDB_CODE_SUCCESS;
-  cJSON* root = NULL;
+  cJSON*    root = NULL;
   SHashObj* keyHash = NULL;
-  int32_t size = 0;
+  int32_t   size = 0;
   // set json NULL data
   if (!json || strtrim((char*)json) == 0 || strcasecmp(json, TSDB_DATA_NULL_STR_L) == 0) {
     retCode = TSDB_CODE_SUCCESS;
@@ -371,7 +371,8 @@ int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag **ppTag, SMs
     }
     STagVal val = {0};
     val.pKey = jsonKey;
-    taosHashPut(keyHash, jsonKey, keyLen, &keyLen, CHAR_BYTES);  // add key to hash to remove dumplicate, value is useless
+    taosHashPut(keyHash, jsonKey, keyLen, &keyLen,
+                CHAR_BYTES);  // add key to hash to remove dumplicate, value is useless
 
     if (item->type == cJSON_String) {  // add json value  format: type|data
       char*   jsonValue = item->valuestring;
@@ -382,8 +383,7 @@ int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag **ppTag, SMs
         goto end;
       }
       val.type = TSDB_DATA_TYPE_NCHAR;
-      if (valLen > 0 && !taosMbsToUcs4(jsonValue, valLen, (TdUcs4*)tmp,
-                                       (int32_t)(valLen * TSDB_NCHAR_SIZE), &valLen)) {
+      if (valLen > 0 && !taosMbsToUcs4(jsonValue, valLen, (TdUcs4*)tmp, (int32_t)(valLen * TSDB_NCHAR_SIZE), &valLen)) {
         uError("charset:%s to %s. val:%s, errno:%s, convert failed.", DEFAULT_UNICODE_ENCODEC, tsCharset, jsonValue,
                strerror(errno));
         retCode = buildSyntaxErrMsg(pMsgBuf, "charset convert json error", jsonValue);
@@ -413,7 +413,7 @@ int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag **ppTag, SMs
 
 end:
   taosHashCleanup(keyHash);
-  if(retCode == TSDB_CODE_SUCCESS){
+  if (retCode == TSDB_CODE_SUCCESS) {
     tTagNew(pTagVals, 1, true, ppTag);
   }
   cJSON_Delete(root);
@@ -679,6 +679,7 @@ int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, S
   tNameExtractFullName(pName, fullName);
   STableMeta** pRes = taosHashGet(pMetaCache->pTableMeta, fullName, strlen(fullName));
   if (NULL == pRes || NULL == *pRes) {
+    parserError("getTableMetaFromCache error: %s", fullName);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   *pMeta = tableMetaDup(*pRes);
@@ -707,6 +708,7 @@ int32_t reserveDbVgInfoInCache(int32_t acctId, const char* pDb, SParseMetaCache*
 int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo) {
   SArray** pRes = taosHashGet(pMetaCache->pDbVgroup, pDbFName, strlen(pDbFName));
   if (NULL == pRes) {
+    parserError("getDbVgInfoFromCache error: %s", pDbFName);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   // *pRes is null, which is a legal value, indicating that the user DB has not been created
@@ -734,6 +736,7 @@ int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName,
   tNameExtractFullName(pName, fullName);
   SVgroupInfo** pRes = taosHashGet(pMetaCache->pTableVgroup, fullName, strlen(fullName));
   if (NULL == pRes || NULL == *pRes) {
+    parserError("getTableVgroupFromCache error: %s", fullName);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   memcpy(pVgroup, *pRes, sizeof(SVgroupInfo));
@@ -748,6 +751,7 @@ int32_t getDbVgVersionFromCache(SParseMetaCache* pMetaCache, const char* pDbFNam
                                 int32_t* pTableNum) {
   SDbInfo** pRes = taosHashGet(pMetaCache->pDbCfg, pDbFName, strlen(pDbFName));
   if (NULL == pRes || NULL == *pRes) {
+    parserError("getDbVgVersionFromCache error: %s", pDbFName);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   *pVersion = (*pRes)->vgVer;
@@ -763,6 +767,7 @@ int32_t reserveDbCfgInCache(int32_t acctId, const char* pDb, SParseMetaCache* pM
 int32_t getDbCfgFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SDbCfgInfo* pInfo) {
   SDbCfgInfo** pRes = taosHashGet(pMetaCache->pDbCfg, pDbFName, strlen(pDbFName));
   if (NULL == pRes || NULL == *pRes) {
+    parserError("getDbCfgFromCache error: %s", pDbFName);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   memcpy(pInfo, *pRes, sizeof(SDbCfgInfo));
@@ -801,6 +806,7 @@ int32_t getUserAuthFromCache(SParseMetaCache* pMetaCache, const char* pUser, con
   int32_t len = userAuthToStringExt(pUser, pDbFName, type, key);
   bool*   pRes = taosHashGet(pMetaCache->pUserAuth, key, len);
   if (NULL == pRes) {
+    parserError("getUserAuthFromCache error: %s, %s, %d", pUser, pDbFName, type);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   *pPass = *pRes;
@@ -820,6 +826,7 @@ int32_t reserveUdfInCache(const char* pFunc, SParseMetaCache* pMetaCache) {
 int32_t getUdfInfoFromCache(SParseMetaCache* pMetaCache, const char* pFunc, SFuncInfo* pInfo) {
   SFuncInfo** pRes = taosHashGet(pMetaCache->pUdf, pFunc, strlen(pFunc));
   if (NULL == pRes || NULL == *pRes) {
+    parserError("getUdfInfoFromCache error: %s", pFunc);
     return TSDB_CODE_PAR_INTERNAL_ERROR;
   }
   memcpy(pInfo, *pRes, sizeof(SFuncInfo));
diff --git a/source/libs/parser/test/parInsertTest.cpp b/source/libs/parser/test/parInsertTest.cpp
index 4d313fca76..eae83e2223 100644
--- a/source/libs/parser/test/parInsertTest.cpp
+++ b/source/libs/parser/test/parInsertTest.cpp
@@ -59,6 +59,7 @@ class InsertTest : public Test {
   }
 
   int32_t runAsync() {
+    cxt_.async = true;
     code_ = parseInsertSyntax(&cxt_, &res_);
     if (code_ != TSDB_CODE_SUCCESS) {
       cout << "parseInsertSyntax code:" << toString(code_) << ", msg:" << errMagBuf_ << endl;

From 338f6eea718ad6fea4114b599fcbd0b3cab788b7 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 5 Jun 2022 00:47:03 +0800
Subject: [PATCH 3/8] fix(query): fix invalid free in exchange.

---
 source/libs/executor/src/executorimpl.c | 43 +++++++++++--------------
 1 file changed, 19 insertions(+), 24 deletions(-)

diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 0cfeecb28c..1a73e9f3ac 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -2418,6 +2418,9 @@ int32_t loadRemoteDataCallback(void* param, const SDataBuf* pMsg, int32_t code)
     pRsp->compLen = htonl(pRsp->compLen);
     pRsp->numOfCols = htonl(pRsp->numOfCols);
     pRsp->useconds = htobe64(pRsp->useconds);
+
+    ASSERT(pSourceDataInfo->pRsp != NULL);
+    qDebug("fetch rsp received, index:%d, rows:%d", pSourceDataInfo->index, pRsp->numOfRows);
   } else {
     pSourceDataInfo->code = code;
   }
@@ -2466,6 +2469,8 @@ static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInf
   SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, sourceIndex);
   SSourceDataInfo*       pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, sourceIndex);
 
+  ASSERT(pDataInfo->status == EX_SOURCE_DATA_NOT_READY);
+
   qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", %d/%" PRIzu, GET_TASKID(pTaskInfo),
          pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId, sourceIndex, totalSources);
 
@@ -2643,7 +2648,6 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
     int32_t completed = 0;
     for (int32_t i = 0; i < totalSources; ++i) {
       SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i);
-
       if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
         completed += 1;
         continue;
@@ -2653,6 +2657,11 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
         continue;
       }
 
+      if (pDataInfo->code != TSDB_CODE_SUCCESS) {
+        code = pDataInfo->code;
+        goto _error;
+      }
+
       SRetrieveTableRsp*     pRsp = pDataInfo->pRsp;
       SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, i);
 
@@ -2690,6 +2699,8 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
                pLoadInfo->totalSize);
       }
 
+      taosMemoryFreeClear(pDataInfo->pRsp);
+
       if (pDataInfo->status != EX_SOURCE_DATA_EXHAUSTED) {
         pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
         code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
@@ -2699,7 +2710,6 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
         }
       }
 
-      taosMemoryFreeClear(pDataInfo->pRsp);
       return pExchangeInfo->pResult;
     }
 
@@ -2719,26 +2729,9 @@ static SSDataBlock* concurrentlyLoadRemoteData(SOperatorInfo* pOperator) {
 
   if (pOperator->status == OP_RES_TO_RETURN) {
     return concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
+  } else {
+    ASSERT(0);
   }
-
-  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSources);
-  int64_t startTs = taosGetTimestampUs();
-
-  // Asynchronously send all fetch requests to all sources.
-  for (int32_t i = 0; i < totalSources; ++i) {
-    int32_t code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
-    if (code != TSDB_CODE_SUCCESS) {
-      return NULL;
-    }
-  }
-
-  int64_t endTs = taosGetTimestampUs();
-  qDebug("%s send all fetch request to %" PRIzu " sources completed, elapsed:%" PRId64, GET_TASKID(pTaskInfo),
-         totalSources, endTs - startTs);
-
-  tsem_wait(&pExchangeInfo->ready);
-  pOperator->status = OP_RES_TO_RETURN;
-  return concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
 }
 
 static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
@@ -2758,10 +2751,11 @@ static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
   }
 
   int64_t endTs = taosGetTimestampUs();
-  qDebug("%s send all fetch request to %" PRIzu " sources completed, elapsed:%" PRId64, GET_TASKID(pTaskInfo),
+  qDebug("%s send all fetch requests to %" PRIzu " sources completed, elapsed:%" PRId64, GET_TASKID(pTaskInfo),
          totalSources, endTs - startTs);
 
   tsem_wait(&pExchangeInfo->ready);
+  pOperator->status = OP_RES_TO_RETURN;
   pOperator->cost.openCost = taosGetTimestampUs() - startTs;
 
   return TSDB_CODE_SUCCESS;
@@ -2875,7 +2869,8 @@ static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) {
   if (pExchangeInfo->seqLoadData) {
     return seqLoadRemoteData(pOperator);
   } else {
-    return concurrentlyLoadRemoteData(pOperator);
+    return concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
+//    return concurrentlyLoadRemoteData(pOperator);
   }
 }
 
@@ -2927,7 +2922,7 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, const SNodeList* p
   }
 
   pInfo->pResult = pBlock;
-  pInfo->seqLoadData = true;
+  pInfo->seqLoadData = false;
 
   tsem_init(&pInfo->ready, 0, 0);
 

From 6fcbcabdfc14ed3701332ac3b1406d9f3ef576c7 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 5 Jun 2022 14:48:15 +0800
Subject: [PATCH 4/8] refactor: do some internal refactor.

---
 source/libs/executor/inc/executorimpl.h |  13 ++-
 source/libs/executor/src/executil.c     |  41 --------
 source/libs/executor/src/executorimpl.c | 127 ++++++------------------
 source/libs/executor/src/scanoperator.c |  92 +++++++++--------
 source/libs/executor/src/sortoperator.c |  15 +++
 5 files changed, 105 insertions(+), 183 deletions(-)

diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index 936e9307d1..061bdc7236 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -727,7 +727,11 @@ SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pR
                                    char* pData, int16_t bytes, bool masterscan, uint64_t groupId,
                                    SExecTaskInfo* pTaskInfo, bool isIntervalQuery, SAggSupporter* pSup);
 
+SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo);
+
 SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, tsdbReaderT pDataReader, SReadHandle* pHandle, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode *pScanPhyNode, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SExprInfo* pScalarExprInfo,
                                            int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo);
@@ -738,9 +742,6 @@ SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pR
 
 SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataBlock* pResBlock, const SName* pName,
-                                              SNode* pCondition, SEpSet epset, SArray* colList,
-                                              SExecTaskInfo* pTaskInfo, bool showRewrite, int32_t accountId);
 SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                           SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
                                           STimeWindowAggSupp *pTwAggSupp, SExecTaskInfo* pTaskInfo);
@@ -758,9 +759,8 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
                                        SExprInfo* pScalarExprInfo, int32_t numOfScalarExpr, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHandle,
-    SArray* pTableIdList, STableScanPhysiNode* pTableScanNode, SExecTaskInfo* pTaskInfo,
-    STimeWindowAggSupp* pTwSup);
+SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHandle, SArray* pTableIdList,
+    STableScanPhysiNode* pTableScanNode, SExecTaskInfo* pTaskInfo, STimeWindowAggSupp* pTwSup);
 
 
 SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
@@ -776,7 +776,6 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo*
                                            SSDataBlock* pResultBlock, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SNode* pOnCondition, SExecTaskInfo* pTaskInfo);
-SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, SExprInfo* pExpr, int32_t numOfOutput, SSDataBlock* pResBlock, SArray* pColMatchInfo, STableListInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream,
     SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap,
diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c
index 1c45e38b63..484ff35e3a 100644
--- a/source/libs/executor/src/executil.c
+++ b/source/libs/executor/src/executil.c
@@ -28,18 +28,6 @@ typedef struct SCompSupporter {
   int32_t           order;
 } SCompSupporter;
 
-int32_t getRowNumForMultioutput(STaskAttr* pQueryAttr, bool topBottomQuery, bool stable) {
-  if (pQueryAttr && (!stable)) {
-    for (int16_t i = 0; i < pQueryAttr->numOfOutput; ++i) {
-//      if (pQueryAttr->pExpr1[i].base. == FUNCTION_TOP || pQueryAttr->pExpr1[i].base.functionId == FUNCTION_BOTTOM) {
-//        return (int32_t)pQueryAttr->pExpr1[i].base.param[0].i;
-//      }
-    }
-  }
-
-  return 1;
-}
-
 int32_t getOutputInterResultBufSize(STaskAttr* pQueryAttr) {
   int32_t size = 0;
 
@@ -113,35 +101,6 @@ void closeResultRow(SResultRow* pResultRow) {
   pResultRow->closed = true;
 }
 
-void clearResultRow(STaskRuntimeEnv *pRuntimeEnv, SResultRow *pResultRow) {
-  if (pResultRow == NULL) {
-    return;
-  }
-
-  // the result does not put into the SDiskbasedBuf, ignore it.
-  if (pResultRow->pageId >= 0) {
-    SFilePage *page = getBufPage(pRuntimeEnv->pResultBuf, pResultRow->pageId);
-
-    int16_t offset = 0;
-    for (int32_t i = 0; i < pRuntimeEnv->pQueryAttr->numOfOutput; ++i) {
-      struct SResultRowEntryInfo *pEntryInfo = NULL;//pResultRow->pEntryInfo[i];
-
-//      int16_t size = pRuntimeEnv->pQueryAttr->pExpr1[i].base.resSchema.bytes;
-//      char * s = getPosInResultPage(pRuntimeEnv->pQueryAttr, page, pResultRow->offset, offset);
-//      memset(s, 0, size);
-
-//      offset += size;
-      cleanupResultRowEntry(pEntryInfo);
-    }
-  }
-
-  pResultRow->numOfRows = 0;
-  pResultRow->pageId = -1;
-  pResultRow->offset = -1;
-  pResultRow->closed = false;
-  pResultRow->win = TSWINDOW_INITIALIZER;
-}
-
 // TODO refactor: use macro
 SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, const int32_t* offset) {
   assert(index >= 0 && offset != NULL);
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 1a73e9f3ac..6434e3a016 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -2896,52 +2896,55 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo) {
   return TSDB_CODE_SUCCESS;
 }
 
-SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, const SNodeList* pSources, SSDataBlock* pBlock,
-                                          SExecTaskInfo* pTaskInfo) {
+static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* pInfo) {
+  size_t numOfSources = LIST_LENGTH(pExNode->pSrcEndPoints);
+
+  pInfo->pSources = taosArrayInit(numOfSources, sizeof(SDownstreamSourceNode));
+  pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo));
+  if (pInfo->pSourceDataInfo == NULL || pInfo->pSources == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  for (int32_t i = 0; i < numOfSources; ++i) {
+    SNodeListNode* pNode = nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
+    taosArrayPush(pInfo->pSources, pNode);
+  }
+
+  return initDataSource(numOfSources, pInfo);
+}
+
+SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo) {
   SExchangeInfo* pInfo = taosMemoryCalloc(1, sizeof(SExchangeInfo));
   SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
 
-  size_t numOfSources = LIST_LENGTH(pSources);
-  pInfo->pSources = taosArrayInit(numOfSources, sizeof(SDownstreamSourceNode));
-  pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo));
-  if (pInfo->pSourceDataInfo == NULL || pInfo->pSources == NULL) {
-    goto _error;
-  }
-
-  for (int32_t i = 0; i < numOfSources; ++i) {
-    SNodeListNode* pNode = nodesListGetNode((SNodeList*)pSources, i);
-    taosArrayPush(pInfo->pSources, pNode);
-  }
-
-  int32_t code = initDataSource(numOfSources, pInfo);
+  int32_t code = initExchangeOperator(pExNode, pInfo);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
 
-  pInfo->pResult = pBlock;
-  pInfo->seqLoadData = false;
-
+  pInfo->seqLoadData  = false;
+  pInfo->pTransporter = pTransporter;
+  pInfo->pResult      = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
   tsem_init(&pInfo->ready, 0, 0);
 
-  pOperator->name = "ExchangeOperator";
+  pOperator->name         = "ExchangeOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;
-  pOperator->blocking = false;
-  pOperator->status = OP_NOT_OPENED;
-  pOperator->info = pInfo;
-  pOperator->numOfExprs = pBlock->info.numOfCols;
-  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->blocking     = false;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->info         = pInfo;
+  pOperator->numOfExprs   = pInfo->pResult->info.numOfCols;
+  pOperator->pTaskInfo    = pTaskInfo;
 
   pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, NULL,
                                          destroyExchangeOperatorInfo, NULL, NULL, NULL);
-  pInfo->pTransporter = pTransporter;
   return pOperator;
 
 _error:
   if (pInfo != NULL) {
-    destroyExchangeOperatorInfo(pInfo, numOfSources);
+    destroyExchangeOperatorInfo(pInfo, LIST_LENGTH(pExNode->pSrcEndPoints));
   }
 
   taosMemoryFreeClear(pInfo);
@@ -4394,9 +4397,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
 
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
-      SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pPhyNode;
-      SSDataBlock*        pResBlock = createResDataBlock(pExchange->node.pOutputDataBlockDesc);
-      return createExchangeOperatorInfo(pHandle->pMsgCb->clientRpc, pExchange->pSrcEndPoints, pResBlock, pTaskInfo);
+      return createExchangeOperatorInfo(pHandle->pMsgCb->clientRpc, (SExchangePhysiNode*)pPhyNode, pTaskInfo);
     } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
       SScanPhysiNode*      pScanPhyNode = (SScanPhysiNode*)pPhyNode;  // simple child table.
       STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
@@ -4424,41 +4425,16 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
       return pOperator;
     } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
       SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode;
-      SScanPhysiNode*            pScanNode = &pSysScanPhyNode->scan;
-
-      SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
-
-      SSDataBlock* pResBlock = createResDataBlock(pDescNode);
-
-      int32_t numOfOutputCols = 0;
-      SArray* colList =
-          extractColMatchInfo(pScanNode->pScanCols, pDescNode, &numOfOutputCols, pTaskInfo, COL_MATCH_FROM_COL_ID);
-      SOperatorInfo* pOperator = createSysTableScanOperatorInfo(
-          pHandle, pResBlock, &pScanNode->tableName, pScanNode->node.pConditions, pSysScanPhyNode->mgmtEpSet, colList,
-          pTaskInfo, pSysScanPhyNode->showRewrite, pSysScanPhyNode->accountId);
-      return pOperator;
+      return createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pTaskInfo);
     } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
       STagScanPhysiNode* pScanPhyNode = (STagScanPhysiNode*)pPhyNode;
 
-      SDataBlockDescNode* pDescNode = pScanPhyNode->node.pOutputDataBlockDesc;
-
-      SSDataBlock* pResBlock = createResDataBlock(pDescNode);
-
       int32_t code = getTableList(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableListInfo, pTagCond);
       if (code != TSDB_CODE_SUCCESS) {
         return NULL;
       }
 
-      int32_t    num = 0;
-      SExprInfo* pExprInfo = createExprInfo(pScanPhyNode->pScanPseudoCols, NULL, &num);
-
-      int32_t numOfOutputCols = 0;
-      SArray* colList = extractColMatchInfo(pScanPhyNode->pScanPseudoCols, pDescNode, &numOfOutputCols, pTaskInfo,
-                                            COL_MATCH_FROM_COL_ID);
-
-      SOperatorInfo* pOperator =
-          createTagScanOperatorInfo(pHandle, pExprInfo, num, pResBlock, colList, pTableListInfo, pTaskInfo);
-      return pOperator;
+      return createTagScanOperatorInfo(pHandle, pScanPhyNode, pTableListInfo, pTaskInfo);
     } else {
       ASSERT(0);
     }
@@ -5016,45 +4992,6 @@ _complete:
   return code;
 }
 
-void setResultBufSize(STaskAttr* pQueryAttr, SResultInfo* pResultInfo) {
-  const int32_t DEFAULT_RESULT_MSG_SIZE = 1024 * (1024 + 512);
-
-  // the minimum number of rows for projection query
-  const int32_t MIN_ROWS_FOR_PRJ_QUERY = 8192;
-  const int32_t DEFAULT_MIN_ROWS = 4096;
-
-  const float THRESHOLD_RATIO = 0.85f;
-
-  //  if (isProjQuery(pQueryAttr)) {
-  //    int32_t numOfRes = DEFAULT_RESULT_MSG_SIZE / pQueryAttr->resultRowSize;
-  //    if (numOfRes < MIN_ROWS_FOR_PRJ_QUERY) {
-  //      numOfRes = MIN_ROWS_FOR_PRJ_QUERY;
-  //    }
-  //
-  //    pResultInfo->capacity = numOfRes;
-  //  } else {  // in case of non-prj query, a smaller output buffer will be used.
-  //    pResultInfo->capacity = DEFAULT_MIN_ROWS;
-  //  }
-
-  pResultInfo->threshold = (int32_t)(pResultInfo->capacity * THRESHOLD_RATIO);
-  pResultInfo->totalRows = 0;
-}
-
-// TODO refactor
-void freeColumnFilterInfo(SColumnFilterInfo* pFilter, int32_t numOfFilters) {
-  if (pFilter == NULL || numOfFilters == 0) {
-    return;
-  }
-
-  for (int32_t i = 0; i < numOfFilters; i++) {
-    if (pFilter[i].filterstr && pFilter[i].pz) {
-      taosMemoryFree((void*)(pFilter[i].pz));
-    }
-  }
-
-  taosMemoryFree(pFilter);
-}
-
 static void doDestroyTableList(STableListInfo* pTableqinfoList) {
   taosArrayDestroy(pTableqinfoList->pTableList);
   taosHashCleanup(pTableqinfoList->map);
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 1633d2f4e6..8541a0b129 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -36,11 +36,6 @@
 #define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN)
 #define SWITCH_ORDER(n)              (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC))
 
-typedef struct SWindowPosition {
-  int32_t pageId;
-  int32_t rowId;
-} SWindowPosition;
-
 static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity);
 static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
                                      const char* dbName);
@@ -1430,48 +1425,59 @@ int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbT
   return numOfRows;
 }
 
-SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pResBlock, const SName* pName,
-                                              SNode* pCondition, SEpSet epset, SArray* colList,
-                                              SExecTaskInfo* pTaskInfo, bool showRewrite, int32_t accountId) {
+SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNode *pScanPhyNode, SExecTaskInfo* pTaskInfo) {
   SSysTableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SSysTableScanInfo));
   SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
-    taosMemoryFreeClear(pInfo);
-    taosMemoryFreeClear(pOperator);
-    terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
-    return NULL;
+    goto _error;
   }
 
-  pInfo->accountId = accountId;
-  pInfo->showRewrite = showRewrite;
-  pInfo->pRes = pResBlock;
-  pInfo->pCondition = pCondition;
-  pInfo->scanCols = colList;
+  SScanPhysiNode* pScanNode = &pScanPhyNode->scan;
+
+  SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
+  SSDataBlock* pResBlock = createResDataBlock(pDescNode);
+
+  int32_t num = 0;
+  SArray* colList = extractColMatchInfo(pScanNode->pScanCols, pDescNode, &num, pTaskInfo, COL_MATCH_FROM_COL_ID);
+
+  pInfo->accountId   = pScanPhyNode->accountId;
+  pInfo->showRewrite = pScanPhyNode->showRewrite;
+  pInfo->pRes        = pResBlock;
+  pInfo->pCondition  = pScanNode->node.pConditions;
+  pInfo->scanCols    = colList;
 
   initResultSizeInfo(pOperator, 4096);
 
-  tNameAssign(&pInfo->name, pName);
+  tNameAssign(&pInfo->name, &pScanNode->tableName);
   const char* name = tNameGetTableName(&pInfo->name);
+
   if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) {
     pInfo->readHandle = *(SReadHandle*)readHandle;
     blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
   } else {
     tsem_init(&pInfo->ready, 0, 0);
-    pInfo->epSet = epset;
+    pInfo->epSet = pScanPhyNode->mgmtEpSet;
     pInfo->readHandle = *(SReadHandle*)readHandle;
   }
 
-  pOperator->name = "SysTableScanOperator";
+  pOperator->name         = "SysTableScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN;
-  pOperator->blocking = false;
-  pOperator->status = OP_NOT_OPENED;
-  pOperator->info = pInfo;
-  pOperator->numOfExprs = pResBlock->info.numOfCols;
+  pOperator->blocking     = false;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->info         = pInfo;
+  pOperator->numOfExprs   = pResBlock->info.numOfCols;
+  pOperator->pTaskInfo    = pTaskInfo;
+
   pOperator->fpSet =
       createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator, NULL, NULL, NULL);
-  pOperator->pTaskInfo = pTaskInfo;
 
   return pOperator;
+
+  _error:
+  taosMemoryFreeClear(pInfo);
+  taosMemoryFreeClear(pOperator);
+  terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
+  return NULL;
 }
 
 static SSDataBlock* doTagScan(SOperatorInfo* pOperator) {
@@ -1626,28 +1632,34 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
   pInfo->pRes = blockDataDestroy(pInfo->pRes);
 }
 
-SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, SExprInfo* pExpr, int32_t numOfOutput,
-                                         SSDataBlock* pResBlock, SArray* pColMatchInfo, STableListInfo* pTableListInfo,
-                                         SExecTaskInfo* pTaskInfo) {
+SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pPhyNode, STableListInfo* pTableListInfo, SExecTaskInfo* pTaskInfo) {
   STagScanInfo*  pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
   SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
 
-  pInfo->pTableList = pTableListInfo;
-  pInfo->pColMatchInfo = pColMatchInfo;
-  pInfo->pRes = pResBlock;
-  pInfo->readHandle = *pReadHandle;
-  pInfo->curPos = 0;
-  pOperator->name = "TagScanOperator";
+  SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;
+
+  int32_t    numOfExprs = 0;
+  SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);
+
+  int32_t num = 0;
+  SArray* colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, pTaskInfo, COL_MATCH_FROM_COL_ID);
+
+  pInfo->pTableList       = pTableListInfo;
+  pInfo->pColMatchInfo    = colList;
+  pInfo->pRes             = createResDataBlock(pDescNode);;
+  pInfo->readHandle       = *pReadHandle;
+  pInfo->curPos           = 0;
+  pOperator->name         = "TagScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
-  pOperator->blocking = false;
-  pOperator->status = OP_NOT_OPENED;
-  pOperator->info = pInfo;
-  pOperator->pExpr = pExpr;
-  pOperator->numOfExprs = numOfOutput;
-  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->blocking     = false;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->info         = pInfo;
+  pOperator->pExpr        = pExprInfo;
+  pOperator->numOfExprs   = numOfExprs;
+  pOperator->pTaskInfo    = pTaskInfo;
 
   initResultSizeInfo(pOperator, 4096);
   blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c
index 8f5fa88070..5e1b47cc91 100644
--- a/source/libs/executor/src/sortoperator.c
+++ b/source/libs/executor/src/sortoperator.c
@@ -1,3 +1,18 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
 #include "tdatablock.h"
 #include "executorimpl.h"
 

From 7df1727a7b4bb701d32e4f7a8323c88eb0ffa8ee Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 5 Jun 2022 15:24:57 +0800
Subject: [PATCH 5/8] refactor(query): do some internal refactor.

---
 source/libs/executor/inc/executorimpl.h | 26 +++----
 source/libs/executor/src/executorimpl.c |  6 +-
 source/libs/executor/src/scanoperator.c | 97 +++++++++++++++----------
 3 files changed, 76 insertions(+), 53 deletions(-)

diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h
index 061bdc7236..c6220b5ffe 100644
--- a/source/libs/executor/inc/executorimpl.h
+++ b/source/libs/executor/inc/executorimpl.h
@@ -303,13 +303,17 @@ typedef struct SScanInfo {
   int32_t numOfDesc;
 } SScanInfo;
 
+typedef struct SSampleExecInfo {
+  double          sampleRatio;  // data block sample ratio, 1 by default
+  uint32_t        seed;         // random seed value
+} SSampleExecInfo;
+
 typedef struct STableScanInfo {
   void*           dataReader;
   SReadHandle     readHandle;
 
   SFileBlockLoadRecorder readRecorder;
   int64_t         numOfRows;
-  int64_t         elapsedTime;
 //  int32_t         prevGroupId;  // previous table group id
   SScanInfo       scanInfo;
   int32_t         scanTimes;
@@ -330,9 +334,8 @@ typedef struct STableScanInfo {
   SQueryTableDataCond cond;
   int32_t         scanFlag;     // table scan flag to denote if it is a repeat/reverse/main scan
   int32_t         dataBlockLoadFlag;
-  double          sampleRatio;  // data block sample ratio, 1 by default
   SInterval       interval;     // if the upstream is an interval operator, the interval info is also kept here to get the time window to check if current data block needs to be loaded.
-
+  SSampleExecInfo sample;       // sample execution info
   int32_t         curTWinIdx;
 } STableScanInfo;
 
@@ -692,7 +695,7 @@ void    doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWin
 int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char* pData, int16_t type, int16_t bytes,
                                 int32_t groupId, SDiskbasedBuf* pBuf, SExecTaskInfo* pTaskInfo, SAggSupporter* pAggSup);
 void    doDestroyBasicInfo(SOptrBasicInfo* pInfo, int32_t numOfOutput);
-int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
+int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
                                   int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
                                   SArray* pColList);
 void    getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key, STimeWindow* win);
@@ -789,8 +792,6 @@ int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBloc
 
 void setInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol);
 
-void copyTsColoum(SSDataBlock* pRes, SqlFunctionCtx* pCtx, int32_t numOfOutput);
-
 bool    isTaskKilled(SExecTaskInfo* pTaskInfo);
 int32_t checkForQueryBuf(size_t numOfTables);
 
@@ -825,16 +826,14 @@ int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SExplainExecInfo
 int32_t aggDecodeResultRow(SOperatorInfo* pOperator, char* result);
 int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length);
 
-STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts,
-                                       SInterval* pInterval, int32_t precision, STimeWindow* win);
-int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn,
-    int32_t startPos, TSKEY ekey, __block_search_fn_t searchFn, STableQueryInfo* item,
-    int32_t order);
+STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
+                                int32_t precision, STimeWindow* win);
+int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
+    __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order);
 int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order);
 int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, const char* pKey);
 SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int64_t tableGroupId, int32_t interBufSize);
-SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap,
-    int32_t* pIndex);
+SResultWindowInfo* getSessionTimeWindow(SArray* pWinInfos, TSKEY ts, int64_t gap, int32_t* pIndex);
 int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t rows,
     int32_t start, int64_t gap, SHashObj* pStDeleted);
 bool functionNeedToExecute(SqlFunctionCtx* pCtx);
@@ -842,6 +841,7 @@ int64_t getSmaWaterMark(int64_t interval, double filesFactor);
 bool isSmaStream(int8_t triggerType);
 
 int32_t compareTimeWindow(const void* p1, const void* p2, const void* param);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 6434e3a016..6a464de689 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -2524,7 +2524,7 @@ void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray
   }
 }
 
-int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
+int32_t setDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadInfo, int32_t numOfRows, char* pData,
                                   int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
                                   SArray* pColList) {
   if (pColList == NULL) {  // data from other sources
@@ -2679,7 +2679,7 @@ static SSDataBlock* concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SEx
       }
 
       SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp;
-      code = setSDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
+      code = setDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
                                        pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL);
       if (code != 0) {
         taosMemoryFreeClear(pDataInfo->pRsp);
@@ -2803,7 +2803,7 @@ static SSDataBlock* seqLoadRemoteData(SOperatorInfo* pOperator) {
     SSDataBlock*       pRes = pExchangeInfo->pResult;
     SRetrieveTableRsp* pTableRsp = pDataInfo->pRsp;
     int32_t            code =
-        setSDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
+        setDataBlockFromFetchRsp(pExchangeInfo->pResult, pLoadInfo, pTableRsp->numOfRows, pTableRsp->data,
                                   pTableRsp->compLen, pTableRsp->numOfCols, startTs, &pDataInfo->totalRows, NULL);
 
     if (pRsp->completed == 1) {
diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index 8541a0b129..ae6924eaff 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -40,6 +40,22 @@ static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capac
 static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
                                      const char* dbName);
 
+static void addTagPseudoColumnData(SReadHandle *pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr, SSDataBlock* pBlock);
+static bool processBlockWithProbability(const SSampleExecInfo *pInfo);
+
+bool processBlockWithProbability(const SSampleExecInfo *pInfo) {
+#if 0
+  if (pInfo->sampleRatio == 1) {
+    return true;
+  }
+
+  uint32_t val = taosRandR((uint32_t*) &pInfo->seed);
+  return (val % ((uint32_t)(1/pInfo->sampleRatio))) == 0;
+#else
+  return true;
+#endif
+}
+
 static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
   for (int32_t i = 0; i < numOfOutput; ++i) {
     SWITCH_ORDER(pCtx[i].order);
@@ -153,8 +169,6 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn
   return false;
 }
 
-static void addTagPseudoColumnData(SReadHandle *pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr, SSDataBlock* pBlock);
-
 static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
                              uint32_t* status) {
   SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
@@ -274,12 +288,9 @@ static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunction
     STimeWindow* pTWindow = &pTableScanInfo->cond.twindows[i];
     TSWAP(pTWindow->skey, pTWindow->ekey);
   }
-  SQueryTableDataCond *pCond = &pTableScanInfo->cond;
-  taosqsort(pCond->twindows,
-            pCond->numOfTWindows,
-            sizeof(STimeWindow),
-            pCond,
-            compareTimeWindow);
+
+  SQueryTableDataCond* pCond = &pTableScanInfo->cond;
+  taosqsort(pCond->twindows, pCond->numOfTWindows, sizeof(STimeWindow), pCond, compareTimeWindow);
 }
 
 void addTagPseudoColumnData(SReadHandle *pHandle, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr, SSDataBlock* pBlock) {
@@ -310,18 +321,19 @@ void addTagPseudoColumnData(SReadHandle *pHandle, SExprInfo* pPseudoExpr, int32_
     } else {  // these are tags
       STagVal tagVal = {0};
       tagVal.cid = pExpr->base.pParam[0].pCol->colId;
-      const char *p = metaGetTableTagVal(&mr.me, pColInfoData->info.type, &tagVal);
+      const char* p = metaGetTableTagVal(&mr.me, pColInfoData->info.type, &tagVal);
 
-      char *data = NULL;
-      if(pColInfoData->info.type != TSDB_DATA_TYPE_JSON && p != NULL){
-        data = tTagValToData((const STagVal *)p, false);
-      }else {
+      char* data = NULL;
+      if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON && p != NULL) {
+        data = tTagValToData((const STagVal*)p, false);
+      } else {
         data = (char*)p;
       }
 
       for (int32_t i = 0; i < pBlock->info.rows; ++i) {
         colDataAppend(pColInfoData, i, data, (data == NULL));
       }
+
       if (data && (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) && p != NULL &&
           IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {
         taosMemoryFree(data);
@@ -359,6 +371,12 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
       longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
     }
 
+    // process this data block based on the probabilities
+    bool processThisBlock = processBlockWithProbability(&pTableScanInfo->sample);
+    if (!processThisBlock) {
+      continue;
+    }
+
     tsdbRetrieveDataBlockInfo(pTableScanInfo->dataReader, &pBlock->info);
 
     uint32_t status = 0;
@@ -433,6 +451,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
       STimeWindow* pWin = &pTableScanInfo->cond.twindows[i];
       qDebug("%s\t qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey);
     }
+
     while (pTableScanInfo->scanTimes < total) {
       while (pTableScanInfo->curTWinIdx < pTableScanInfo->cond.numOfTWindows) {
         SSDataBlock* p = doTableScanImpl(pOperator);
@@ -502,11 +521,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo));
   SOperatorInfo*  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
   if (pInfo == NULL || pOperator == NULL) {
-    taosMemoryFreeClear(pInfo);
-    taosMemoryFreeClear(pOperator);
-
-    pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
-    return NULL;
+    goto _error;
   }
 
   SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
@@ -517,7 +532,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
 
   int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
   if (code != TSDB_CODE_SUCCESS) {
-    return NULL;
+    goto _error;
   }
 
   if (pTableScanNode->scan.pScanPseudoCols != NULL) {
@@ -528,32 +543,40 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   pInfo->scanInfo = (SScanInfo){.numOfAsc = pTableScanNode->scanSeq[0], .numOfDesc = pTableScanNode->scanSeq[1]};
 //    pInfo->scanInfo = (SScanInfo){.numOfAsc = 0, .numOfDesc = 1}; // for debug purpose
 
-  pInfo->readHandle = *readHandle;
-  pInfo->interval = extractIntervalInfo(pTableScanNode);
-  pInfo->sampleRatio = pTableScanNode->ratio;
-  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
-  pInfo->pResBlock = createResDataBlock(pDescNode);
-  pInfo->pFilterNode = pTableScanNode->scan.node.pConditions;
-  pInfo->dataReader = pDataReader;
-  pInfo->scanFlag = MAIN_SCAN;
-  pInfo->pColMatchInfo = pColList;
-  pInfo->curTWinIdx = 0;
+  pInfo->readHandle        = *readHandle;
+  pInfo->interval          = extractIntervalInfo(pTableScanNode);
+  pInfo->sample.sampleRatio= pTableScanNode->ratio;
+  pInfo->sample.seed       = taosGetTimestampMs();
 
-  pOperator->name = "TableScanOperator";  // for debug purpose
+  pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
+  pInfo->pResBlock         = createResDataBlock(pDescNode);
+  pInfo->pFilterNode       = pTableScanNode->scan.node.pConditions;
+  pInfo->dataReader        = pDataReader;
+  pInfo->scanFlag          = MAIN_SCAN;
+  pInfo->pColMatchInfo     = pColList;
+  pInfo->curTWinIdx        = 0;
+
+  pOperator->name         = "TableScanOperator";  // for debug purpose
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
-  pOperator->blocking = false;
-  pOperator->status = OP_NOT_OPENED;
-  pOperator->info = pInfo;
-  pOperator->numOfExprs = numOfCols;
-  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->blocking     = false;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->info         = pInfo;
+  pOperator->numOfExprs   = numOfCols;
+  pOperator->pTaskInfo    = pTaskInfo;
 
   pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo,
                                          NULL, NULL, getTableScannerExecInfo);
 
   // for non-blocking operator, the open cost is always 0
   pOperator->cost.openCost = 0;
-
   return pOperator;
+
+  _error:
+  taosMemoryFreeClear(pInfo);
+  taosMemoryFreeClear(pOperator);
+
+  pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
+  return NULL;
 }
 
 SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) {
@@ -1351,7 +1374,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
       }
 
       SRetrieveMetaTableRsp* pTableRsp = pInfo->pRsp;
-      setSDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pTableRsp->numOfRows, pTableRsp->data,
+      setDataBlockFromFetchRsp(pInfo->pRes, &pInfo->loadInfo, pTableRsp->numOfRows, pTableRsp->data,
                                 pTableRsp->compLen, pOperator->numOfExprs, startTs, NULL, pInfo->scanCols);
 
       // todo log the filter info

From 6bd6822637b56adb631833018e418779a5fbd3da Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 5 Jun 2022 15:26:25 +0800
Subject: [PATCH 6/8] refactor: remove unused functions.

---
 source/libs/executor/src/executorimpl.c | 11 -----------
 1 file changed, 11 deletions(-)

diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 6a464de689..c8f49d5dd0 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -2723,17 +2723,6 @@ _error:
   return NULL;
 }
 
-static SSDataBlock* concurrentlyLoadRemoteData(SOperatorInfo* pOperator) {
-  SExchangeInfo* pExchangeInfo = pOperator->info;
-  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
-
-  if (pOperator->status == OP_RES_TO_RETURN) {
-    return concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
-  } else {
-    ASSERT(0);
-  }
-}
-
 static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
   SExchangeInfo* pExchangeInfo = pOperator->info;
   SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

From e25560b64086d05704a663ae6d437664800e9aa8 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 5 Jun 2022 15:35:34 +0800
Subject: [PATCH 7/8] refactor: do some internal refactor.

---
 source/libs/executor/src/scanoperator.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c
index ae6924eaff..eadfd9be88 100644
--- a/source/libs/executor/src/scanoperator.c
+++ b/source/libs/executor/src/scanoperator.c
@@ -546,7 +546,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
   pInfo->readHandle        = *readHandle;
   pInfo->interval          = extractIntervalInfo(pTableScanNode);
   pInfo->sample.sampleRatio= pTableScanNode->ratio;
-  pInfo->sample.seed       = taosGetTimestampMs();
+  pInfo->sample.seed       = taosGetTimestampSec();
 
   pInfo->dataBlockLoadFlag = pTableScanNode->dataRequired;
   pInfo->pResBlock         = createResDataBlock(pDescNode);

From 4391ed4af6666992dfbf6adf323931a6ad0b322e Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Sun, 5 Jun 2022 15:37:04 +0800
Subject: [PATCH 8/8] test:update the test suit.

---
 tests/script/tsim/testsuit.sim | 64 +++++++++++++++++-----------------
 1 file changed, 32 insertions(+), 32 deletions(-)

diff --git a/tests/script/tsim/testsuit.sim b/tests/script/tsim/testsuit.sim
index b2ca1e0345..c73efde9d5 100644
--- a/tests/script/tsim/testsuit.sim
+++ b/tests/script/tsim/testsuit.sim
@@ -3,8 +3,8 @@
 #run tsim/user/privilege2.sim
 #run tsim/user/user_len.sim
 #run tsim/user/privilege1.sim
-##run tsim/user/pass_len.sim
-##run tsim/table/basic1.sim
+#run tsim/user/pass_len.sim
+#run tsim/table/basic1.sim
 #run tsim/trans/lossdata1.sim
 #run tsim/trans/create_db.sim
 ##run tsim/stable/alter_metrics.sim
@@ -13,20 +13,20 @@
 ##run tsim/stable/column_drop.sim
 ##run tsim/stable/column_modify.sim
 ##run tsim/stable/tag_rename.sim
-##run tsim/stable/vnode3.sim
-##run tsim/stable/metrics.sim
+#run tsim/stable/vnode3.sim
+#run tsim/stable/metrics.sim
 ##run tsim/stable/alter_insert2.sim
-##run tsim/stable/show.sim
-##run tsim/stable/alter_import.sim
+#run tsim/stable/show.sim
+#run tsim/stable/alter_import.sim
 ##run tsim/stable/tag_add.sim
-##run tsim/stable/tag_drop.sim
-##run tsim/stable/column_add.sim
-##run tsim/stable/alter_count.sim
-##run tsim/stable/values.sim
-##run tsim/stable/dnode3.sim
-##run tsim/stable/alter_insert1.sim
-##run tsim/stable/refcount.sim
-##run tsim/stable/disk.sim
+#run tsim/stable/tag_drop.sim
+#run tsim/stable/column_add.sim
+#run tsim/stable/alter_count.sim
+#run tsim/stable/values.sim
+#run tsim/stable/dnode3.sim
+#run tsim/stable/alter_insert1.sim
+#run tsim/stable/refcount.sim
+#run tsim/stable/disk.sim
 #run tsim/db/basic1.sim
 #run tsim/db/basic3.sim
 ##run tsim/db/basic7.sim
@@ -40,24 +40,24 @@
 #run tsim/mnode/basic3.sim
 #run tsim/mnode/basic2.sim
 #run tsim/parser/fourArithmetic-basic.sim
-run tsim/parser/groupby-basic.sim
-run tsim/snode/basic1.sim
-run tsim/query/time_process.sim
-run tsim/query/stddev.sim
-run tsim/query/interval-offset.sim
-run tsim/query/charScalarFunction.sim
-run tsim/query/complex_select.sim
-run tsim/query/explain.sim
-run tsim/query/crash_sql.sim
-run tsim/query/diff.sim
-run tsim/query/complex_limit.sim
-run tsim/query/complex_having.sim
-run tsim/query/udf.sim
-run tsim/query/complex_group.sim
-run tsim/query/interval.sim
-run tsim/query/session.sim
-run tsim/query/scalarFunction.sim
-run tsim/query/scalarNull.sim
+#run tsim/parser/groupby-basic.sim
+#run tsim/snode/basic1.sim
+#run tsim/query/time_process.sim
+#run tsim/query/stddev.sim
+#run tsim/query/interval-offset.sim
+#run tsim/query/charScalarFunction.sim
+#run tsim/query/complex_select.sim
+#run tsim/query/explain.sim
+#run tsim/query/crash_sql.sim
+#run tsim/query/diff.sim
+#run tsim/query/complex_limit.sim
+#run tsim/query/complex_having.sim
+#run tsim/query/udf.sim
+#run tsim/query/complex_group.sim
+#run tsim/query/interval.sim
+#run tsim/query/session.sim
+#run tsim/query/scalarFunction.sim
+##run tsim/query/scalarNull.sim
 run tsim/query/complex_where.sim
 run tsim/tmq/basic1.sim
 run tsim/tmq/basic4.sim