fix(stream): add resetting tagScan logic & fix parse value error if value is integer & fix tmqError.py (#34124)

Author: wangmm0220

source/libs/executor/src/executor.c

@@ -350,7 +350,8 @@ bool qNeedReset(qTaskInfo_t pInfo) {
   int32_t node = nodeType(pOperator->pPhyNode);
   return (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == node || 
           QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == node ||
-          QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == node);
+          QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == node ||
+          QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == node);
 }
 
 static void setReadHandle(SReadHandle* pHandle, STableScanBase* pScanBaseInfo) {

source/libs/executor/src/scanoperator.c

@@ -3845,12 +3845,6 @@ static int32_t doTagScanFromMetaEntryNext(SOperatorInfo* pOperator, SSDataBlock*
   return code;
 }
 
-static SSDataBlock* doTagScanFromMetaEntry(SOperatorInfo* pOperator) {
-  SSDataBlock* pRes = NULL;
-  int32_t      code = doTagScanFromMetaEntryNext(pOperator, &pRes);
-  return pRes;
-}
-
 static void destroyTagScanOperatorInfo(void* param) {
   STagScanInfo* pInfo = (STagScanInfo*)param;
   if (pInfo->pCtbCursor != NULL && pInfo->pStorageAPI != NULL) {
@@ -3870,6 +3864,24 @@ static void destroyTagScanOperatorInfo(void* param) {
   taosMemoryFreeClear(param);
 }
 
+static int32_t resetTagScanOperatorState(SOperatorInfo* pOper) {
+  int32_t         code = TSDB_CODE_SUCCESS;
+  STagScanInfo*   pInfo = pOper->info;
+
+  pOper->status = OP_NOT_OPENED;
+  STagScanPhysiNode* pTagScanNode = (STagScanPhysiNode*)pOper->pPhyNode;
+  if (pTagScanNode->onlyMetaCtbIdx) {
+      SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;
+      SStorageAPI*   pAPI = &pTaskInfo->storageAPI;
+      pAPI->metaFn.closeCtbCursor(pInfo->pCtbCursor);
+      pInfo->pCtbCursor = NULL;
+  } else {
+      pInfo->curPos = 0;
+  }
+
+  return code;
+}
+
 int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* pTagScanNode,
                                   STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
                                   SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
@@ -3884,7 +3896,7 @@ int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* p
     code = terrno;
     goto _error;
   }
-
+  pOperator->pPhyNode = pTagScanNode;
   SDataBlockDescNode* pDescNode = pPhyNode->node.pOutputDataBlockDesc;
 
   int32_t    numOfExprs = 0;
@@ -3940,6 +3952,8 @@ int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* p
   __optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;
   pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,
                                          optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
+  setOperatorResetStateFn(pOperator, resetTagScanOperatorState);     
+  
   *pOptrInfo = pOperator;
   return code;
 

source/libs/parser/src/parAstCreater.c

@@ -659,8 +659,7 @@ SNode* createValueNode(SAstCreateContext* pCxt, int32_t dataType, const SToken*
   }
   copyValueTrimEscape(val->literal, pLiteral->n + 1, pLiteral,
                       pCxt->pQueryCxt->hasDupQuoteChar && (TK_NK_ID == pLiteral->type));
-  if (TK_NK_ID != pLiteral->type && TK_TIMEZONE != pLiteral->type &&
-      (IS_VAR_DATA_TYPE(dataType) || TSDB_DATA_TYPE_TIMESTAMP == dataType)) {
+  if (TK_NK_STRING == pLiteral->type) {
     (void)trimString(pLiteral->z, pLiteral->n, val->literal, pLiteral->n);
   }
   val->node.resType.type = dataType;

test/cases/17-DataSubscription/02-Consume/test_tmq_error.py

@@ -235,6 +235,7 @@ def tmqCase1(self, cfgPath, buildPath):
         tmqCom.getStartConsumeNotifyFromTmqsim()
         tdLog.info("================= stop dnode, and remove data file, then start dnode ===========================")
         tdDnodes.stop(1)
+        tmqCom.stopTmqSimProcess("tmq_sim")
 
         time.sleep(5)
         dataPath = buildPath + "/../sim/dnode1/data/*"