Merge branch 'fix/stmt_rerun' into enh/stopquery_a

source/client/src/clientStmt.c

@@ -357,6 +357,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.inExecCache = true;
 
     if (pStmt->sql.autoCreateTbl) {
+      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
       return TSDB_CODE_SUCCESS;
     }
   }
@@ -365,9 +366,11 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     if (pStmt->bInfo.inExecCache) {
       ASSERT(taosHashGetSize(pStmt->exec.pBlockHash) == 1);
       pStmt->bInfo.needParse = false;
+      tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
       return TSDB_CODE_SUCCESS;
     }
 
+    tscDebug("no stmt block cache for tb %s", pStmt->bInfo.tbFName);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -391,6 +394,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
         STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
       }
 
+      tscDebug("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64 , pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
+
       return TSDB_CODE_SUCCESS;
     }
 
@@ -406,6 +411,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
   if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
     STMT_ERR_RET(stmtCleanBindInfo(pStmt));
 
+    tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
+
     return TSDB_CODE_SUCCESS;
   }
 
@@ -420,6 +427,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
   if (uid == pStmt->bInfo.tbUid) {
     pStmt->bInfo.needParse = false;
 
+    tscDebug("tb %s is current table", pStmt->bInfo.tbFName);
+
     return TSDB_CODE_SUCCESS;
   }
 
@@ -440,6 +449,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.boundTags = pCache->boundTags;
     pStmt->bInfo.tagsCached = true;
 
+    tscDebug("tb %s in execBlock list, set to current", pStmt->bInfo.tbFName);
+
     return TSDB_CODE_SUCCESS;
   }
 
@@ -461,6 +472,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
       STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
     }
 
+    tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
+
     return TSDB_CODE_SUCCESS;
   }
 
@@ -510,6 +523,8 @@ TAOS_STMT* stmtInit(STscObj* taos) {
 int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
+  tscDebug("stmt start to prepare");
+
   if (pStmt->sql.status >= STMT_PREPARE) {
     STMT_ERR_RET(stmtResetStmt(pStmt));
   }
@@ -529,6 +544,8 @@ int stmtPrepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
 int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
+  tscDebug("stmt start to set tbName: %s", tbName);
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
 
   int32_t insert = 0;
@@ -559,6 +576,8 @@ int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
 int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
+  tscDebug("stmt start to set tbTags");
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
 
   if (pStmt->bInfo.inExecCache) {
@@ -572,6 +591,7 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
     STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
   }
 
+  tscDebug("start to bind stmt tag values");
   STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.sname.tname,
                                   tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
 
@@ -617,6 +637,8 @@ int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields
 int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
+  tscDebug("start to bind stmt data, colIdx: %d", colIdx);
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
 
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
@@ -707,6 +729,8 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
 int stmtAddBatch(TAOS_STMT* stmt) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
+  tscDebug("stmt start to add batch");
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
 
   STMT_ERR_RET(stmtCacheBlock(pStmt));
@@ -715,6 +739,8 @@ int stmtAddBatch(TAOS_STMT* stmt) {
 }
 
 int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
+  tscDebug("stmt start to update tbUid, blockNum: %d", pRsp->nBlocks);
+
   if (pRsp->nBlocks <= 0) {
     tscError("invalid submit resp block number %d", pRsp->nBlocks);
     STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
@@ -727,11 +753,6 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
     char*             key = taosHashGetKey(pIter, &keyLen);
 
     STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
-    if (pMeta->uid != pStmt->bInfo.tbUid) {
-      tscError("table uid %" PRIx64 " mis-match with current table uid %" PRIx64, pMeta->uid, pStmt->bInfo.tbUid);
-      STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
-    }
-
     if (pMeta->uid) {
       pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
       continue;
@@ -775,6 +796,8 @@ int stmtExec(TAOS_STMT* stmt) {
   SSubmitRsp* pRsp = NULL;
   bool        autoCreateTbl = pStmt->exec.autoCreateTbl;
 
+  tscDebug("stmt start to exec");
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
 
   if (STMT_TYPE_QUERY == pStmt->sql.type) {

source/libs/parser/inc/parInsertData.h

@@ -20,6 +20,7 @@
 #include "os.h"
 #include "tname.h"
 #include "ttypes.h"
+#include "query.h"
 
 #define IS_DATA_COL_ORDERED(spd) ((spd->orderStatus) == (int8_t)ORDER_STATUS_ORDERED)
 

source/libs/parser/src/parInsertData.c

@@ -630,6 +630,17 @@ int32_t qCloneStmtDataBlock(void** pDst, void* pSrc) {
   memcpy(*pDst, pSrc, sizeof(STableDataBlocks));
   ((STableDataBlocks*)(*pDst))->cloned = true;
 
+  STableDataBlocks* pBlock = (STableDataBlocks*)(*pDst);
+  if (pBlock->pTableMeta) {
+    void *pNewMeta = taosMemoryMalloc(TABLE_META_SIZE(pBlock->pTableMeta));
+    if (NULL == pNewMeta) {
+      taosMemoryFreeClear(*pDst);
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    memcpy(pNewMeta, pBlock->pTableMeta, TABLE_META_SIZE(pBlock->pTableMeta));
+    pBlock->pTableMeta = pNewMeta;
+  }
+
   return qResetStmtDataBlock(*pDst, false);
 }
 

source/libs/planner/src/planPhysiCreater.c

@@ -1519,6 +1519,7 @@ static SSubplan* makeSubplan(SPhysiPlanContext* pCxt, SLogicSubplan* pLogicSubpl
 
 static int32_t buildInsertValuesSubplan(SPhysiPlanContext* pCxt, SVnodeModifyLogicNode* pModify, SSubplan* pSubplan) {
   pSubplan->msgType = pModify->msgType;
+  pSubplan->execNode.nodeId = pModify->pVgDataBlocks->vg.vgId;
   pSubplan->execNode.epSet = pModify->pVgDataBlocks->vg.epSet;
   return createDataInserter(pCxt, pModify->pVgDataBlocks, &pSubplan->pDataSink);
 }