Merge pull request #9676 from taosdata/feature/3.0_liaohj

Feature/3.0 liaohj

include/libs/catalog/catalog.h

@@ -87,14 +87,14 @@ int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName,
 /**
  * Get a DB's all vgroup info.
  * @param pCatalog (input, got with catalogGetHandle)
- * @param pRpc (input, rpc object)
+ * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
  * @param pDBName (input, full db name)
  * @param forceUpdate (input, force update db vgroup info from mnode) 
  * @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
  * @return error code
  */
-int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
+int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const char* pDBName, bool forceUpdate, SArray** pVgroupList);
 
 int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo);
 
@@ -149,13 +149,13 @@ int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, con
 /**
  * Get a table's actual vgroup, for stable it's all possible vgroup list.
  * @param pCatalog (input, got with catalogGetHandle)
- * @param pRpc (input, rpc object)
+ * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
  * @param pTableName (input, table name, NOT including db name)
  * @param pVgroupList (output, vgroup info list, element is SVgroupInfo, NEED to simply free the array by caller)
  * @return error code
  */
-int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
+int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList);
 
 /**
  * Get a table's vgroup from its name's hash value.
@@ -172,16 +172,16 @@ int32_t catalogGetTableHashVgroup(struct SCatalog* pCatalog, void * pTransporter
 /**
  * Get all meta data required in pReq.
  * @param pCatalog (input, got with catalogGetHandle)
- * @param pRpc (input, rpc object)
+ * @param pTransporter (input, rpc object)
  * @param pMgmtEps (input, mnode EPs)
  * @param pReq (input, reqest info)
  * @param pRsp (output, response data)
  * @return error code 
  */
-int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
+int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp);
 
 
-int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList);
+int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, SArray* pQnodeList);
 
 int32_t catalogGetExpiredSTables(struct SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num);
 

include/libs/parser/parser.h

@@ -24,7 +24,6 @@ extern "C" {
 
 typedef struct SParseContext {
   SParseBasicCtx   ctx;
-  int8_t           schemaAttached; // denote if submit block is built with table schema or not
   const char      *pSql;           // sql string
   size_t           sqlLen;         // length of the sql string
   char            *pMsg;           // extended error message if exists to help identifying the problem in sql statement.
@@ -41,8 +40,17 @@ typedef struct SParseContext {
  */
 int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQuery);
 
-bool qIsDdlQuery(const SQueryNode* pQuery);
+/**
+ * Return true if it is a ddl/dcl sql statement
+ * @param pQuery
+ * @return
+ */
+bool qIsDdlQuery(const SQueryNode* pQueryNode);
 
+/**
+ * Destroy logic query plan
+ * @param pQueryNode
+ */
 void qDestroyQuery(SQueryNode* pQueryNode);
 
 /**
@@ -62,8 +70,8 @@ void columnListDestroy(SArray* pColumnList);
 void dropAllExprInfo(SArray** pExprInfo, int32_t numOfLevel);
 
 typedef struct SSourceParam {
-  SArray            *pExprNodeList; //Array<struct tExprNode*>
-  SArray            *pColumnList;   //Array<struct SColumn>
+  SArray    *pExprNodeList; //Array<struct tExprNode*>
+  SArray    *pColumnList;   //Array<struct SColumn>
   int32_t    num;
 } SSourceParam;
 

include/libs/planner/planner.h

@@ -139,9 +139,13 @@ typedef struct SQueryDag {
 
 struct SQueryNode;
 
-/**
- * Create the physical plan for the query, according to the AST.
- */
+ /**
+  * Create the physical plan for the query, according to the AST.
+  * @param pQueryInfo
+  * @param pDag
+  * @param requestId
+  * @return
+  */
 int32_t qCreateQueryDag(const struct SQueryNode* pQueryInfo, struct SQueryDag** pDag, uint64_t requestId);
 
 // Set datasource of this subplan, multiple calls may be made to a subplan.

include/libs/qcom/query.h

@@ -87,7 +87,7 @@ typedef struct SUseDbOutput {
   SDBVgroupInfo dbVgroup;
 } SUseDbOutput;
 
-typedef enum {
+enum {
   META_TYPE_NON_TABLE = 1,
   META_TYPE_CTABLE,
   META_TYPE_TABLE,

include/libs/scheduler/scheduler.h

@@ -75,6 +75,12 @@ int32_t scheduleExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void
  */
 int32_t scheduleAsyncExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, void** pJob);
 
+/**
+ * Fetch query result from the remote query executor
+ * @param pJob
+ * @param data
+ * @return
+ */
 int32_t scheduleFetchRows(void *pJob, void **data);
 
 
@@ -85,6 +91,10 @@ int32_t scheduleFetchRows(void *pJob, void **data);
  */
 int32_t scheduleCancelJob(void *pJob);
 
+/**
+ * Free the query job
+ * @param pJob
+ */
 void scheduleFreeJob(void *pJob);
 
 void schedulerDestroy(void);

source/client/inc/clientInt.h

@@ -62,6 +62,7 @@ typedef struct SAppInstInfo {
   SList            *pConnList;  // STscObj linked list
   int64_t           clusterId;
   void             *pTransporter;
+  SHeartBeatInfo hb;
 } SAppInstInfo;
 
 typedef struct SAppInfo {
@@ -70,7 +71,7 @@ typedef struct SAppInfo {
   char          *ep;
   int32_t        pid;
   int32_t        numOfThreads;
-  SHeartBeatInfo hb;
+
   SHashObj      *pInstMap;
 } SAppInfo;
 

source/client/src/clientImpl.c

@@ -291,10 +291,10 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
 
   nPrintTsc("%s", sql)
 
-  SRequestObj* pRequest = NULL;
-  SQueryNode* pQuery = NULL;
-  SQueryDag* pDag = NULL;
-  void* pJob = NULL;
+  SRequestObj *pRequest = NULL;
+  SQueryNode  *pQuery   = NULL;
+  SQueryDag   *pDag     = NULL;
+  void        *pJob     = NULL;
 
   terrno = TSDB_CODE_SUCCESS;
   CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);

source/client/src/clientMain.c

 #include "os.h"
+#include "tref.h"
+#include "trpc.h"
 #include "clientInt.h"
 #include "clientLog.h"
 #include "query.h"
 #include "tmsg.h"
 #include "tglobal.h"
-#include "tref.h"
-#include "trpc.h"
+#include "catalog.h"
 
 #define TSC_VAR_NOT_RELEASE 1
 #define TSC_VAR_RELEASED    0

source/client/test/clientTests.cpp

@@ -48,7 +48,7 @@ int main(int argc, char** argv) {
 }
 
 TEST(testCase, driverInit_Test) { taos_init(); }
-
+#if 0
 TEST(testCase, connect_Test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   if (pConn == NULL) {
@@ -485,6 +485,22 @@ TEST(testCase, drop_stable_Test) {
   taos_close(pConn);
 }
 
+TEST(testCase, generated_request_id_test) {
+  SHashObj* phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
+
+  for (int32_t i = 0; i < 50000; ++i) {
+    uint64_t v = generateRequestId();
+    void*    result = taosHashGet(phash, &v, sizeof(v));
+    if (result != nullptr) {
+      printf("0x%lx, index:%d\n", v, i);
+    }
+    assert(result == nullptr);
+    taosHashPut(phash, &v, sizeof(v), NULL, 0);
+  }
+
+  taosHashCleanup(phash);
+}
+
 // TEST(testCase, create_topic_Test) {
 //  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
 //  assert(pConn != NULL);
@@ -518,46 +534,55 @@ TEST(testCase, drop_stable_Test) {
 //  tmq_create_topic(pConn, "test_topic_1", sql, strlen(sql));
 //  taos_close(pConn);
 //}
-TEST(testCase, generated_request_id_test) {
-  SHashObj* phash = taosHashInit(10000, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
-
-  for (int32_t i = 0; i < 50000; ++i) {
-    uint64_t v = generateRequestId();
-    void*    result = taosHashGet(phash, &v, sizeof(v));
-    if (result != nullptr) {
-      printf("0x%lx, index:%d\n", v, i);
-    }
-    assert(result == nullptr);
-    taosHashPut(phash, &v, sizeof(v), NULL, 0);
-  }
-
-  taosHashCleanup(phash);
-}
-
-// TEST(testCase, projection_query_tables) {
+//TEST(testCase, insert_test) {
 //  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
 //  ASSERT_EQ(pConn, nullptr);
 //
 //  TAOS_RES* pRes = taos_query(pConn, "use abc1");
 //  taos_free_result(pRes);
 //
-//  pRes = taos_query(pConn, "select * from t_2");
+//  pRes = taos_query(pConn, "insert into t_2 values(now, 1)");
 //  if (taos_errno(pRes) != 0) {
 //    printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
 //    taos_free_result(pRes);
 //    ASSERT_TRUE(false);
 //  }
 //
-//  TAOS_ROW pRow = NULL;
-//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-//  int32_t numOfFields = taos_num_fields(pRes);
-//
-//  char str[512] = {0};
-//  while((pRow = taos_fetch_row(pRes)) != NULL) {
-//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
-//    printf("%s\n", str);
-//  }
-//
 //  taos_free_result(pRes);
 //  taos_close(pConn);
 //}
+#endif
+
+TEST(testCase, projection_query_tables) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  TAOS_RES* pRes = taos_query(pConn, "use test1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to use db, reason:%s", taos_errstr(pRes));
+    taos_free_result(pRes);
+    return;
+  }
+
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "select * from tm0");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create multiple tables, reason:%s\n", taos_errstr(pRes));
+    taos_free_result(pRes);
+    ASSERT_TRUE(false);
+  }
+
+  TAOS_ROW    pRow = NULL;
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  int32_t     numOfFields = taos_num_fields(pRes);
+
+  char str[512] = {0};
+  while ((pRow = taos_fetch_row(pRes)) != NULL) {
+    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+    printf("%s\n", str);
+  }
+
+  taos_free_result(pRes);
+  taos_close(pConn);
+}

source/libs/catalog/src/catalog.c

@@ -1288,8 +1288,8 @@ _return:
 }
 
 
-int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp) {
-  if (NULL == pCatalog || NULL == pRpc  || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
+int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp) {
+  if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
     CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
@@ -1312,7 +1312,7 @@ int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* p
       SName *name = taosArrayGet(pReq->pTableName, i);
       STableMeta *pTableMeta = NULL;
       
-      CTG_ERR_JRET(ctgGetTableMeta(pCatalog, pRpc, pMgmtEps, name, false, &pTableMeta, -1));
+      CTG_ERR_JRET(ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, name, false, &pTableMeta, -1));
 
       if (NULL == taosArrayPush(pRsp->pTableMeta, &pTableMeta)) {
         ctgError("taosArrayPush failed, idx:%d", i);
@@ -1325,7 +1325,6 @@ int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* p
   return TSDB_CODE_SUCCESS;
 
 _return:  
-
   if (pRsp->pTableMeta) {
     int32_t aSize = taosArrayGetSize(pRsp->pTableMeta);
     for (int32_t i = 0; i < aSize; ++i) {

source/libs/parser/inc/parserInt.h

@@ -44,14 +44,14 @@ void clearAllTableMetaInfo(SQueryStmtInfo* pQueryInfo, bool removeMeta, uint64_t
 
 /**
  * Validate the sql info, according to the corresponding metadata info from catalog.
- * @param pCatalog
- * @param pSqlInfo
- * @param pQueryInfo a bounded AST with essential meta data from local buffer or mgmt node
- * @param id
- * @param msg
+ * @param pCtx
+ * @param pInfo
+ * @param pQueryInfo
+ * @param msgBuf
+ * @param msgBufLen
  * @return
  */
-int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pSqlInfo, SQueryStmtInfo* pQueryInfo, int64_t id, char* msg, int32_t msgLen);
+int32_t qParserValidateSqlNode(SParseBasicCtx *pCtx, SSqlInfo* pInfo, SQueryStmtInfo* pQueryInfo, char* msgBuf, int32_t msgBufLen);
 
 /**
  * validate the ddl ast, and convert the ast to the corresponding message format
@@ -62,6 +62,14 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pSqlInfo, SQ
  */
 SDclStmtInfo* qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, char* msgBuf, int32_t msgBufLen);
 
+/**
+ *
+ * @param pInfo
+ * @param pCtx
+ * @param msgBuf
+ * @param msgBufLen
+ * @return
+ */
 SVnodeModifOpStmtInfo* qParserValidateCreateTbSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, char* msgBuf, int32_t msgBufLen);
 
 /**

source/libs/parser/src/astValidate.c

@@ -3630,12 +3630,12 @@ int32_t evaluateSqlNode(SSqlNode* pNode, int32_t tsPrecision, SMsgBuf* pMsgBuf)
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQueryStmtInfo* pQueryInfo, int64_t id, char* msgBuf, int32_t msgBufLen) {
-  assert(pCatalog != NULL && pInfo != NULL);
+int32_t qParserValidateSqlNode(SParseBasicCtx *pCtx, SSqlInfo* pInfo, SQueryStmtInfo* pQueryInfo, char* msgBuf, int32_t msgBufLen) {
+  assert(pCtx != NULL && pInfo != NULL);
   int32_t code = 0;
 
-  SMsgBuf m = {.buf = msgBuf, .len = msgBufLen};
-  SMsgBuf *pMsgBuf = &m;
+  SMsgBuf  m = {.buf = msgBuf, .len = msgBufLen};
+  SMsgBuf* pMsgBuf = &m;
 
   switch (pInfo->type) {
 #if 0
@@ -3682,22 +3682,6 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
       break;
     }
 
-    case TSDB_SQL_USE_DB: {
-      const char* msg = "invalid db name";
-      SToken* pToken = taosArrayGet(pInfo->pMiscInfo->a, 0);
-
-      if (tscValidateName(pToken) != TSDB_CODE_SUCCESS) {
-        return buildInvalidOperationMsg(pMsgBuf, msg);
-      }
-
-      int32_t ret = tNameSetDbName(&pTableMetaInfo->name, getAccountId(pSql), pToken);
-      if (ret != TSDB_CODE_SUCCESS) {
-        return buildInvalidOperationMsg(pMsgBuf, msg);
-      }
-
-      break;
-    }
-
     case TSDB_SQL_RESET_CACHE: {
       return TSDB_CODE_SUCCESS;
     }
@@ -3712,55 +3696,6 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
       break;
     }
 
-    case TSDB_SQL_CREATE_DNODE: {
-      const char* msg = "invalid host name (ip address)";
-
-      if (taosArrayGetSize(pInfo->pMiscInfo->a) > 1) {
-        return buildInvalidOperationMsg(pMsgBuf, msg);
-      }
-
-      SToken* id = taosArrayGet(pInfo->pMiscInfo->a, 0);
-      if (id->type == TK_STRING) {
-        id->n = strdequote(id->z);
-      }
-      break;
-    }
-
-    case TSDB_SQL_CREATE_ACCT:
-    case TSDB_SQL_ALTER_ACCT: {
-      const char* msg1 = "invalid state option, available options[no, r, w, all]";
-      const char* msg2 = "invalid user/account name";
-      const char* msg3 = "name too long";
-
-      SToken* pName = &pInfo->pMiscInfo->user.user;
-      SToken* pPwd = &pInfo->pMiscInfo->user.passwd;
-
-      if (handlePassword(pCmd, pPwd) != TSDB_CODE_SUCCESS) {
-        return TSDB_CODE_TSC_INVALID_OPERATION;
-      }
-
-      if (pName->n >= TSDB_USER_LEN) {
-        return buildInvalidOperationMsg(pMsgBuf, msg3);
-      }
-
-      if (tscValidateName(pName) != TSDB_CODE_SUCCESS) {
-        return buildInvalidOperationMsg(pMsgBuf, msg2);
-      }
-
-      SCreateAcctInfo* pAcctOpt = &pInfo->pMiscInfo->acctOpt;
-      if (pAcctOpt->stat.n > 0) {
-        if (pAcctOpt->stat.z[0] == 'r' && pAcctOpt->stat.n == 1) {
-        } else if (pAcctOpt->stat.z[0] == 'w' && pAcctOpt->stat.n == 1) {
-        } else if (strncmp(pAcctOpt->stat.z, "all", 3) == 0 && pAcctOpt->stat.n == 3) {
-        } else if (strncmp(pAcctOpt->stat.z, "no", 2) == 0 && pAcctOpt->stat.n == 2) {
-        } else {
-          return buildInvalidOperationMsg(pMsgBuf, msg1);
-        }
-      }
-
-      break;
-    }
-
     case TSDB_SQL_DESCRIBE_TABLE: {
       const char* msg1 = "invalid table name";
 
@@ -3865,29 +3800,6 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
       return TSDB_CODE_SUCCESS;
     }
 
-    case TSDB_SQL_CREATE_TABLE: {
-      SCreateTableSql* pCreateTable = pInfo->pCreateTableInfo;
-
-      if (pCreateTable->type == TSQL_CREATE_TABLE || pCreateTable->type == TSQL_CREATE_STABLE) {
-        if ((code = doCheckForCreateTable(pSql, 0, pInfo)) != TSDB_CODE_SUCCESS) {
-          return code;
-        }
-
-      } else if (pCreateTable->type == TSQL_CREATE_TABLE_FROM_STABLE) {
-        assert(pCmd->numOfCols == 0);
-        if ((code = doCheckForCreateFromStable(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
-          return code;
-        }
-
-      } else if (pCreateTable->type == TSQL_CREATE_STREAM) {
-        if ((code = doCheckForStream(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
-          return code;
-        }
-      }
-
-      break;
-    }
-
     case TSDB_SQL_SELECT: {
       const char * msg1 = "no nested query supported in union clause";
       code = loadAllTableMeta(pSql, pInfo);
@@ -3981,13 +3893,14 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
       }
       break;
     }
-    #endif
     default:
       return buildInvalidOperationMsg(pMsgBuf, "not support sql expression");
   }
+#endif
+  }
 
-  SCatalogReq req = {0};
-  SMetaData data = {0};
+  SCatalogReq req  = {0};
+  SMetaData   data = {0};
 
   // TODO: check if the qnode info has been cached already
   req.qNodeRequired = true;
@@ -3997,7 +3910,7 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
   }
 
   // load the meta data from catalog
-  code = catalogGetAllMeta(pCatalog, NULL, NULL, &req, &data);
+  code = catalogGetAllMeta(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, &req, &data);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }

source/libs/parser/src/dCDAstProcess.c

-#include <astGenerator.h>
-#include <tmsg.h>
-#include "astToMsg.h"
+#include "tmsg.h"
+#include "tglobal.h"
 #include "parserInt.h"
+#include "ttime.h"
+#include "astToMsg.h"
+#include "astGenerator.h"
 #include "parserUtil.h"
 #include "queryInfoUtil.h"
-#include "tglobal.h"
-#include "tmsg.h"
-#include "ttime.h"
 
 /* is contained in pFieldList or not */
 static bool has(SArray* pFieldList, int32_t startIndex, const char* name) {

source/libs/parser/src/insertParser.c

@@ -629,7 +629,6 @@ int32_t parseInsertSql(SParseContext* pContext, SVnodeModifOpStmtInfo** pInfo) {
 
   *pInfo = context.pOutput;
   context.pOutput->nodeType = TSDB_SQL_INSERT;
-  context.pOutput->schemaAttache = pContext->schemaAttached;
   context.pOutput->payloadType = PAYLOAD_TYPE_KV;
 
   int32_t code = skipInsertInto(&context);

source/libs/parser/src/parser.c

@@ -31,8 +31,8 @@ bool isInsertSql(const char* pStr, size_t length) {
   } while (1);
 }
 
-bool qIsDdlQuery(const SQueryNode* pQuery) {
-  return TSDB_SQL_INSERT != pQuery->type && TSDB_SQL_SELECT != pQuery->type && TSDB_SQL_CREATE_TABLE != pQuery->type;
+bool qIsDdlQuery(const SQueryNode* pQueryNode) {
+  return TSDB_SQL_INSERT != pQueryNode->type && TSDB_SQL_SELECT != pQueryNode->type && TSDB_SQL_CREATE_TABLE != pQueryNode->type;
 }
 
 int32_t parseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
@@ -67,7 +67,7 @@ int32_t parseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
       return terrno;
     }
 
-    int32_t code = qParserValidateSqlNode(pCxt->ctx.pCatalog, &info, pQueryInfo, pCxt->ctx.requestId, pCxt->pMsg, pCxt->msgLen);
+    int32_t code = qParserValidateSqlNode(&pCxt->ctx, &info, pQueryInfo, pCxt->pMsg, pCxt->msgLen);
     if (code == TSDB_CODE_SUCCESS) {
       *pQuery = (SQueryNode*)pQueryInfo;
     }