feat(tmq): optimize the logic of changing tablelist for stb sub (#35133)

Author: wangmm0220

include/common/tmsg.h

@@ -5597,7 +5597,7 @@ enum {
 };
 
 enum {
-  WITH_DATA = 0,
+  ONLY_DATA = 0,
   WITH_META = 1,
   ONLY_META = 2,
 };
@@ -6203,12 +6203,6 @@ typedef struct {
   int32_t debugFlag;
 } SMqHbRsp;
 
-typedef struct {
-  SMsgHead head;
-  int64_t  consumerId;
-  char     subKey[TSDB_SUBSCRIBE_KEY_LEN];
-} SMqSeekReq;
-
 #define TD_AUTO_CREATE_TABLE 0x1
 typedef struct {
   int64_t       suid;
@@ -6336,9 +6330,6 @@ int32_t tSerializeSMqHbRsp(void* buf, int32_t bufLen, SMqHbRsp* pRsp);
 int32_t tDeserializeSMqHbRsp(void* buf, int32_t bufLen, SMqHbRsp* pRsp);
 void    tDestroySMqHbRsp(SMqHbRsp* pRsp);
 
-int32_t tSerializeSMqSeekReq(void* buf, int32_t bufLen, SMqSeekReq* pReq);
-int32_t tDeserializeSMqSeekReq(void* buf, int32_t bufLen, SMqSeekReq* pReq);
-
 #define TD_REQ_FROM_APP               0x0
 #define SUBMIT_REQ_AUTO_CREATE_TABLE  0x1
 #define SUBMIT_REQ_COLUMN_DATA_FORMAT 0x2

include/common/tmsgdef.h

@@ -466,7 +466,7 @@
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_SUBSCRIBE, "vnode-tmq-subscribe", SMqRebVgReq, SMqRebVgRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_DELETE_SUB, "vnode-tmq-delete-sub", SMqVDeleteReq, SMqVDeleteRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_COMMIT_OFFSET, "vnode-tmq-commit-offset", STqOffset, STqOffset)
-  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_SEEK, "vnode-tmq-seek", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_SEEK, "vnode-tmq-seek", NULL, NULL)      // no longer used
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_ADD_CHECKINFO, "vnode-tmq-add-checkinfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_DEL_CHECKINFO, "vnode-del-checkinfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_CONSUME, "vnode-tmq-consume", SMqPollReq, SMqDataBlkRsp)

include/libs/executor/executor.h

@@ -162,10 +162,10 @@ bool    qTaskIsDone(qTaskInfo_t tinfo);
  */
 int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type);
 
-void    qUpdateTableTagCacheForTmq(qTaskInfo_t tinfo, const SArray* tableIdList, SArray* cids, SArray* cidListArray);
-int32_t qUpdateTableListForTmqScanner(qTaskInfo_t tinfo, const SArray* tableIdList);
-int32_t qDeleteTableListForTmqScanner(qTaskInfo_t tinfo, const SArray* tableIdList);
-int32_t qAddTableListForTmqScanner(qTaskInfo_t tinfo, const SArray* tableIdList);
+void    qUpdateTableTagCacheForQuerySub(qTaskInfo_t tinfo, const SArray* tableIdList, SArray* cids, SArray* cidListArray);
+int32_t qUpdateTableListForQuerySub(qTaskInfo_t tinfo, const SArray* tableIdList);
+int32_t qDeleteTableListForQuerySub(qTaskInfo_t tinfo, const SArray* tableIdList);
+int32_t qAddTableListForQuerySub(qTaskInfo_t tinfo, const SArray* tableIdList);
 
 bool qIsDynamicExecTask(qTaskInfo_t tinfo);
 
@@ -321,7 +321,7 @@ bool    isTaskKilled(void* pTaskInfo);
 
 
 bool    isTrueForSatisfied(STrueForInfo* pTrueForInfo, int64_t skey, int64_t ekey, int64_t count);
-int32_t qFilterTableList(void* pVnode, SArray* uidList, SNode* node, void* pTaskInfo, uint64_t suid);
+int32_t qFilterTableList(void* pVnode, SArray* uidList, int64_t version, SNode* node, void* pTaskInfo, uint64_t suid);
 bool    checkCidInTagCondition(SNode* node, SArray* cidList);
 SNode*  getTagCondNodeForStableTmq(void* node);
 SNode*  getTagCondNodeForQueryTmq(void* tinfo);

include/libs/executor/storageapi.h

@@ -286,7 +286,7 @@ typedef struct SStoreMeta {
   int32_t (*cursorPrev)(SMTbCursor* pTbCur, ETableType jumpTableType);              // metaTbCursorPrev
 
   int32_t (*getTableTags)(void* pVnode, uint64_t suid, SArray* uidList);
-  int32_t (*getTableTagsByUid)(void* pVnode, int64_t suid, SArray* uidList);
+  int32_t (*getTableTagsByUidVersion)(void* pVnode, int64_t suid, SArray* uidList, int64_t version);
   const void* (*extractTagVal)(const void* tag, int16_t type, STagVal* tagVal);  // todo remove it
 
   int32_t (*getTableUidByName)(void* pVnode, char* tbName, uint64_t* uid);

source/client/inc/clientRawBlockWrite.h

@@ -0,0 +1,78 @@
+/*
+ * 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/>.
+ */
+
+#ifndef CLIENT_RAW_BLOCK_WRITE_H
+#define CLIENT_RAW_BLOCK_WRITE_H
+
+#include "clientInt.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// Common macros shared across clientRawBlockWrite.c and clientRawBlockJson.c
+
+#define RAW_LOG_END                                                           \
+  if (code != 0) {                                                            \
+    uError("%s failed at line:%d since:%s", __func__, lino, tstrerror(code)); \
+  } else {                                                                    \
+    uDebug("%s return success", __func__);                                    \
+  }
+
+#define RAW_LOG_START uDebug("%s start", __func__);
+
+#define RAW_NULL_CHECK(c) \
+  do {                    \
+    if (c == NULL) {      \
+      lino = __LINE__;    \
+      code = terrno;      \
+      goto end;           \
+    }                     \
+  } while (0)
+
+#define RAW_FALSE_CHECK(c)           \
+  do {                               \
+    if (!(c)) {                      \
+      code = TSDB_CODE_INVALID_PARA; \
+      lino = __LINE__;               \
+      goto end;                      \
+    }                                \
+  } while (0)
+
+#define RAW_RETURN_CHECK(c) \
+  do {                      \
+    code = c;               \
+    if (code != 0) {        \
+      lino = __LINE__;      \
+      goto end;             \
+    }                       \
+  } while (0)
+
+#define LOG_ID_TAG   "connId:0x%" PRIx64 ", QID:0x%" PRIx64 ",func:%s"
+#define LOG_ID_VALUE *(int64_t*)taos, pRequest->requestId, __func__
+
+#define PROCESS_TABLE_NOT_EXIST(code, tbName) \
+  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) { \
+    uWarn(LOG_ID_TAG " %s not exists, skip", LOG_ID_VALUE, tbName); \
+    code = TSDB_CODE_SUCCESS; \
+    continue; \
+  } \
+  RAW_RETURN_CHECK(code);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // CLIENT_RAW_BLOCK_WRITE_H