feat(tmq): create topic new grammar

example/src/tmq.c

@@ -106,8 +106,8 @@ int32_t create_topic() {
   }
   taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
-  pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");
+  pRes = taos_query(pConn, "create topic topic_ctb_column as database abc1");
+  /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");*/
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/tmsg.h

@@ -2170,9 +2170,9 @@ typedef struct {
   int64_t newConsumerId;
   char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
   int8_t  subType;
-  int8_t  withTbName;
-  int8_t  withSchema;
-  int8_t  withTag;
+  // int8_t  withTbName;
+  // int8_t  withSchema;
+  // int8_t  withTag;
   char* qmsg;
 } SMqRebVgReq;
 
@@ -2184,10 +2184,10 @@ static FORCE_INLINE int32_t tEncodeSMqRebVgReq(void** buf, const SMqRebVgReq* pR
   tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
   tlen += taosEncodeString(buf, pReq->subKey);
   tlen += taosEncodeFixedI8(buf, pReq->subType);
-  tlen += taosEncodeFixedI8(buf, pReq->withTbName);
-  tlen += taosEncodeFixedI8(buf, pReq->withSchema);
-  tlen += taosEncodeFixedI8(buf, pReq->withTag);
-  if (pReq->subType == TOPIC_SUB_TYPE__TABLE) {
+  // tlen += taosEncodeFixedI8(buf, pReq->withTbName);
+  // tlen += taosEncodeFixedI8(buf, pReq->withSchema);
+  // tlen += taosEncodeFixedI8(buf, pReq->withTag);
+  if (pReq->subType == TOPIC_SUB_TYPE__COLUMN) {
     tlen += taosEncodeString(buf, pReq->qmsg);
   }
   return tlen;
@@ -2200,10 +2200,10 @@ static FORCE_INLINE void* tDecodeSMqRebVgReq(const void* buf, SMqRebVgReq* pReq)
   buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
   buf = taosDecodeStringTo(buf, pReq->subKey);
   buf = taosDecodeFixedI8(buf, &pReq->subType);
-  buf = taosDecodeFixedI8(buf, &pReq->withTbName);
-  buf = taosDecodeFixedI8(buf, &pReq->withSchema);
-  buf = taosDecodeFixedI8(buf, &pReq->withTag);
-  if (pReq->subType == TOPIC_SUB_TYPE__TABLE) {
+  // buf = taosDecodeFixedI8(buf, &pReq->withTbName);
+  // buf = taosDecodeFixedI8(buf, &pReq->withSchema);
+  // buf = taosDecodeFixedI8(buf, &pReq->withTag);
+  if (pReq->subType == TOPIC_SUB_TYPE__COLUMN) {
     buf = taosDecodeString(buf, &pReq->qmsg);
   }
   return (void*)buf;

source/dnode/mnode/impl/inc/mndDef.h

@@ -461,10 +461,10 @@ typedef struct {
   int64_t uid;
   int64_t dbUid;
   int32_t version;
-  int8_t         subType;  // db or table
-  int8_t         withTbName;
-  int8_t         withSchema;
-  int8_t         withTag;
+  int8_t  subType;  // column, db or stable
+  // int8_t         withTbName;
+  // int8_t         withSchema;
+  // int8_t         withTag;
   SRWLatch       lock;
   int32_t        consumerCnt;
   int32_t        sqlLen;
@@ -532,9 +532,9 @@ typedef struct {
   int64_t  dbUid;
   int32_t  vgNum;
   int8_t   subType;
-  int8_t    withTbName;
-  int8_t    withSchema;
-  int8_t    withTag;
+  // int8_t    withTbName;
+  // int8_t    withSchema;
+  // int8_t    withTag;
   SHashObj* consumerHash;   // consumerId -> SMqConsumerEp
   SArray*   unassignedVgs;  // SArray<SMqVgEp*>
 } SMqSubscribeObj;

source/dnode/mnode/impl/src/mndDef.c

@@ -396,9 +396,9 @@ SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) {
 
   pSubNew->dbUid = pSub->dbUid;
   pSubNew->subType = pSub->subType;
-  pSubNew->withTbName = pSub->withTbName;
-  pSubNew->withSchema = pSub->withSchema;
-  pSubNew->withTag = pSub->withTag;
+  /*pSubNew->withTbName = pSub->withTbName;*/
+  /*pSubNew->withSchema = pSub->withSchema;*/
+  /*pSubNew->withTag = pSub->withTag;*/
 
   pSubNew->vgNum = pSub->vgNum;
   pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
@@ -431,9 +431,9 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
   tlen += taosEncodeFixedI64(buf, pSub->dbUid);
   tlen += taosEncodeFixedI32(buf, pSub->vgNum);
   tlen += taosEncodeFixedI8(buf, pSub->subType);
-  tlen += taosEncodeFixedI8(buf, pSub->withTbName);
-  tlen += taosEncodeFixedI8(buf, pSub->withSchema);
-  tlen += taosEncodeFixedI8(buf, pSub->withTag);
+  /*tlen += taosEncodeFixedI8(buf, pSub->withTbName);*/
+  /*tlen += taosEncodeFixedI8(buf, pSub->withSchema);*/
+  /*tlen += taosEncodeFixedI8(buf, pSub->withTag);*/
 
   void   *pIter = NULL;
   int32_t sz = taosHashGetSize(pSub->consumerHash);
@@ -458,9 +458,9 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub) {
   buf = taosDecodeFixedI64(buf, &pSub->dbUid);
   buf = taosDecodeFixedI32(buf, &pSub->vgNum);
   buf = taosDecodeFixedI8(buf, &pSub->subType);
-  buf = taosDecodeFixedI8(buf, &pSub->withTbName);
-  buf = taosDecodeFixedI8(buf, &pSub->withSchema);
-  buf = taosDecodeFixedI8(buf, &pSub->withTag);
+  /*buf = taosDecodeFixedI8(buf, &pSub->withTbName);*/
+  /*buf = taosDecodeFixedI8(buf, &pSub->withSchema);*/
+  /*buf = taosDecodeFixedI8(buf, &pSub->withTag);*/
 
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);

source/dnode/mnode/impl/src/mndScheduler.c

@@ -506,7 +506,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
   SQueryPlan* pPlan = NULL;
   SSubplan*   plan = NULL;
 
-  if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
+  if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {
     pPlan = qStringToQueryPlan(pTopic->physicalPlan);
     if (pPlan == NULL) {
       terrno = TSDB_CODE_QRY_INVALID_INPUT;
@@ -552,7 +552,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
 
     mDebug("init subscription %s, assign vg: %d", pSub->key, pVgEp->vgId);
 
-    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
+    if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {
       int32_t msgLen;
 
       plan->execNode.epSet = pVgEp->epSet;

source/dnode/mnode/impl/src/mndSubscribe.c

@@ -93,9 +93,9 @@ static SMqSubscribeObj *mndCreateSub(SMnode *pMnode, const SMqTopicObj *pTopic,
   }
   pSub->dbUid = pTopic->dbUid;
   pSub->subType = pTopic->subType;
-  pSub->withTbName = pTopic->withTbName;
-  pSub->withSchema = pTopic->withSchema;
-  pSub->withTag = pTopic->withTag;
+  /*pSub->withTbName = pTopic->withTbName;*/
+  /*pSub->withSchema = pTopic->withSchema;*/
+  /*pSub->withTag = pTopic->withTag;*/
 
   ASSERT(pSub->unassignedVgs->size == 0);
   ASSERT(taosHashGetSize(pSub->consumerHash) == 0);
@@ -120,9 +120,9 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
   req.vgId = pRebVg->pVgEp->vgId;
   req.qmsg = pRebVg->pVgEp->qmsg;
   req.subType = pSub->subType;
-  req.withTbName = pSub->withTbName;
-  req.withSchema = pSub->withSchema;
-  req.withTag = pSub->withTag;
+  /*req.withTbName = pSub->withTbName;*/
+  /*req.withSchema = pSub->withSchema;*/
+  /*req.withTag = pSub->withTag;*/
   strncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
 
   int32_t tlen = sizeof(SMsgHead) + tEncodeSMqRebVgReq(NULL, &req);

source/dnode/mnode/impl/src/mndTopic.c

@@ -96,9 +96,9 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
   SDB_SET_INT64(pRaw, dataPos, pTopic->dbUid, TOPIC_ENCODE_OVER);
   SDB_SET_INT32(pRaw, dataPos, pTopic->version, TOPIC_ENCODE_OVER);
   SDB_SET_INT8(pRaw, dataPos, pTopic->subType, TOPIC_ENCODE_OVER);
-  SDB_SET_INT8(pRaw, dataPos, pTopic->withTbName, TOPIC_ENCODE_OVER);
-  SDB_SET_INT8(pRaw, dataPos, pTopic->withSchema, TOPIC_ENCODE_OVER);
-  SDB_SET_INT8(pRaw, dataPos, pTopic->withTag, TOPIC_ENCODE_OVER);
+  /*SDB_SET_INT8(pRaw, dataPos, pTopic->withTbName, TOPIC_ENCODE_OVER);*/
+  /*SDB_SET_INT8(pRaw, dataPos, pTopic->withSchema, TOPIC_ENCODE_OVER);*/
+  /*SDB_SET_INT8(pRaw, dataPos, pTopic->withTag, TOPIC_ENCODE_OVER);*/
 
   SDB_SET_INT32(pRaw, dataPos, pTopic->consumerCnt, TOPIC_ENCODE_OVER);
   SDB_SET_INT32(pRaw, dataPos, pTopic->sqlLen, TOPIC_ENCODE_OVER);
@@ -168,9 +168,9 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
   SDB_GET_INT64(pRaw, dataPos, &pTopic->dbUid, TOPIC_DECODE_OVER);
   SDB_GET_INT32(pRaw, dataPos, &pTopic->version, TOPIC_DECODE_OVER);
   SDB_GET_INT8(pRaw, dataPos, &pTopic->subType, TOPIC_DECODE_OVER);
-  SDB_GET_INT8(pRaw, dataPos, &pTopic->withTbName, TOPIC_DECODE_OVER);
-  SDB_GET_INT8(pRaw, dataPos, &pTopic->withSchema, TOPIC_DECODE_OVER);
-  SDB_GET_INT8(pRaw, dataPos, &pTopic->withTag, TOPIC_DECODE_OVER);
+  /*SDB_GET_INT8(pRaw, dataPos, &pTopic->withTbName, TOPIC_DECODE_OVER);*/
+  /*SDB_GET_INT8(pRaw, dataPos, &pTopic->withSchema, TOPIC_DECODE_OVER);*/
+  /*SDB_GET_INT8(pRaw, dataPos, &pTopic->withTag, TOPIC_DECODE_OVER);*/
 
   SDB_GET_INT32(pRaw, dataPos, &pTopic->consumerCnt, TOPIC_DECODE_OVER);
 
@@ -308,11 +308,19 @@ static SDDropTopicReq *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgroup, SMq
 }
 
 static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
-  if (pCreate->name[0] == 0 || pCreate->sql == NULL || pCreate->sql[0] == 0 || pCreate->subDbName[0] == 0) {
   terrno = TSDB_CODE_MND_INVALID_TOPIC;
-    return -1;
+
+  if (pCreate->sql == NULL) return -1;
+
+  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
+    if (pCreate->ast == NULL || pCreate->ast[0] == 0) return -1;
+  } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {
+    if (pCreate->subStbName[0] == 0) return -1;
+  } else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {
+    if (pCreate->subDbName[0] == 0) return -1;
   }
 
+  terrno = TSDB_CODE_SUCCESS;
   return 0;
 }
 
@@ -328,12 +336,11 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
   topicObj.version = 1;
   topicObj.sql = strdup(pCreate->sql);
   topicObj.sqlLen = strlen(pCreate->sql) + 1;
-  /*topicObj.refConsumerCnt = 0;*/
+  topicObj.subType = pCreate->subType;
 
-  if (pCreate->ast && pCreate->ast[0]) {
+  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
     topicObj.ast = strdup(pCreate->ast);
     topicObj.astLen = strlen(pCreate->ast) + 1;
-    topicObj.subType = TOPIC_SUB_TYPE__TABLE;
     /*topicObj.withTbName = pCreate->withTbName;*/
     /*topicObj.withSchema = pCreate->withSchema;*/
 
@@ -368,13 +375,12 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       taosMemoryFree(topicObj.sql);
       return -1;
     }
-  } else {
-    topicObj.ast = NULL;
-    topicObj.astLen = 0;
-    topicObj.physicalPlan = NULL;
-    topicObj.subType = TOPIC_SUB_TYPE__DB;
-    topicObj.withTbName = 1;
-    topicObj.withSchema = 1;
+    /*} else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {*/
+    /*topicObj.ast = NULL;*/
+    /*topicObj.astLen = 0;*/
+    /*topicObj.physicalPlan = NULL;*/
+    /*topicObj.withTbName = 1;*/
+    /*topicObj.withSchema = 1;*/
   }
 
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_TOPIC, pReq);

source/dnode/vnode/CMakeLists.txt

@@ -51,6 +51,7 @@ target_sources(
 
     # tq
     "src/tq/tq.c"
+    "src/tq/tqExec.c"
     "src/tq/tqCommit.c"
     "src/tq/tqOffset.c"
     "src/tq/tqPush.c"

source/dnode/vnode/src/inc/tq.h

@@ -44,21 +44,27 @@ extern "C" {
 typedef struct STqOffsetCfg   STqOffsetCfg;
 typedef struct STqOffsetStore STqOffsetStore;
 
+// tqRead
+
 struct STqReadHandle {
   int64_t           ver;
-  SHashObj*         tbIdHash;
   const SSubmitReq* pMsg;
   SSubmitBlk*       pBlock;
   SSubmitMsgIter    msgIter;
   SSubmitBlkIter    blkIter;
+
   SMeta*    pVnodeMeta;
+  SHashObj* tbIdHash;
   SArray*   pColIdList;  // SArray<int16_t>
-  int32_t           sver;
+
+  int32_t         cachedSchemaVer;
   int64_t         cachedSchemaUid;
   SSchemaWrapper* pSchemaWrapper;
   STSchema*       pSchema;
 };
 
+// tqPush
+
 typedef struct {
   int64_t  consumerId;
   int32_t  epoch;
@@ -68,14 +74,15 @@ typedef struct {
   SRpcMsg* handle;
 } STqPushHandle;
 
+#if 0
 typedef struct {
   char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
   int64_t consumerId;
   int32_t epoch;
   int8_t  subType;
-  int8_t        withTbName;
-  int8_t        withSchema;
-  int8_t        withTag;
+  // int8_t        withTbName;
+  // int8_t        withSchema;
+  // int8_t        withTag;
   char*         qmsg;
   SHashObj*     pDropTbUid;
   STqPushHandle pushHandle;
@@ -85,15 +92,55 @@ typedef struct {
   STqReadHandle* pExecReader[5];
   qTaskInfo_t    task[5];
 } STqExec;
+#endif
+
+// tqExec
+
+typedef struct {
+  char*       qmsg;
+  qTaskInfo_t task[5];
+} STqExecCol;
+
+typedef struct {
+  int64_t suid;
+} STqExecTb;
+
+typedef struct {
+  SHashObj* pFilterOutTbUid;
+} STqExecDb;
+
+typedef struct {
+  int8_t subType;
 
-int32_t tEncodeSTqExec(SEncoder* pEncoder, const STqExec* pExec);
-int32_t tDecodeSTqExec(SDecoder* pDecoder, STqExec* pExec);
+  STqReadHandle* pExecReader[5];
+  union {
+    STqExecCol execCol;
+    STqExecTb  execTb;
+    STqExecDb  execDb;
+  } exec;
+} STqExecHandle;
+
+typedef struct {
+  // info
+  char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
+  int64_t consumerId;
+  int32_t epoch;
+
+  // reader
+  SWalReadHandle* pWalReader;
+
+  // push
+  STqPushHandle pushHandle;
+
+  // exec
+  STqExecHandle execHandle;
+} STqHandle;
 
 struct STQ {
   char*     path;
-  SHashObj* pushMgr;  // consumerId -> STqExec*
-  SHashObj* execs;    // subKey -> STqExec
-  SHashObj* pStreamTasks;
+  SHashObj* pushMgr;       // consumerId -> STqHandle*
+  SHashObj* handles;       // subKey -> STqHandle
+  SHashObj* pStreamTasks;  // taksId -> SStreamTask
   SVnode*   pVnode;
   SWal*     pWal;
   TDB*      pMetaStore;
@@ -111,6 +158,16 @@ static STqMgmt tqMgmt = {0};
 int  tqInit();
 void tqCleanUp();
 
+// int32_t tEncodeSTqExec(SEncoder* pEncoder, const STqExec* pExec);
+// int32_t tDecodeSTqExec(SDecoder* pDecoder, STqExec* pExec);
+
+int32_t tEncodeSTqHandle(SEncoder* pEncoder, const STqHandle* pHandle);
+int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
+
+int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalHead** pHeadWithCkSum);
+
+int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId);
+
 // tqOffset
 STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
 void            STqOffsetClose(STqOffsetStore*);

source/dnode/vnode/src/tq/tqExec.c

+/*
+ * 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 "tq.h"
+
+static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataBlkRsp* pRsp) {
+  int32_t dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
+  void*   buf = taosMemoryCalloc(1, dataStrLen);
+  if (buf == NULL) return -1;
+
+  SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)buf;
+  pRetrieve->useconds = 0;
+  pRetrieve->precision = TSDB_DEFAULT_PRECISION;
+  pRetrieve->compressed = 0;
+  pRetrieve->completed = 1;
+  pRetrieve->numOfRows = htonl(pBlock->info.rows);
+
+  // TODO enable compress
+  int32_t actualLen = 0;
+  blockCompressEncode(pBlock, pRetrieve->data, &actualLen, pBlock->info.numOfCols, false);
+  actualLen += sizeof(SRetrieveTableRsp);
+  ASSERT(actualLen <= dataStrLen);
+  taosArrayPush(pRsp->blockDataLen, &actualLen);
+  taosArrayPush(pRsp->blockData, &buf);
+  return 0;
+}
+
+static int32_t tqAddBlockSchemaToRsp(const STqExecHandle* pExec, int32_t workerId, SMqDataBlkRsp* pRsp) {
+  SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
+  taosArrayPush(pRsp->blockSchema, &pSW);
+  return 0;
+}
+
+static int32_t tqAddTbNameToRsp(const STQ* pTq, const STqExecHandle* pExec, SMqDataBlkRsp* pRsp, int32_t workerId) {
+  SMetaReader mr = {0};
+  metaReaderInit(&mr, pTq->pVnode->pMeta, 0);
+  int64_t uid = pExec->pExecReader[workerId]->msgIter.uid;
+  if (metaGetTableEntryByUid(&mr, uid) < 0) {
+    ASSERT(0);
+    return -1;
+  }
+  char* tbName = strdup(mr.me.name);
+  taosArrayPush(pRsp->blockTbName, &tbName);
+  metaReaderClear(&mr);
+  return 0;
+}
+
+int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkRsp* pRsp, int32_t workerId) {
+  if (pExec->subType == TOPIC_SUB_TYPE__COLUMN) {
+    qTaskInfo_t task = pExec->exec.execCol.task[workerId];
+    ASSERT(task);
+    qSetStreamInput(task, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK, false);
+    while (1) {
+      SSDataBlock* pDataBlock = NULL;
+      uint64_t     ts = 0;
+      if (qExecTask(task, &pDataBlock, &ts) < 0) {
+        ASSERT(0);
+      }
+      if (pDataBlock == NULL) break;
+
+      ASSERT(pDataBlock->info.rows != 0);
+      ASSERT(pDataBlock->info.numOfCols != 0);
+
+      tqAddBlockDataToRsp(pDataBlock, pRsp);
+      if (pRsp->withTbName) {
+        tqAddTbNameToRsp(pTq, pExec, pRsp, workerId);
+      }
+      pRsp->blockNum++;
+    }
+  } else if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
+    pRsp->withSchema = 1;
+    STqReadHandle* pReader = pExec->pExecReader[workerId];
+    tqReadHandleSetMsg(pReader, pReq, 0);
+    while (tqNextDataBlock(pReader)) {
+      SSDataBlock block = {0};
+      if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
+                              &block.info.numOfCols) < 0) {
+        if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;
+        ASSERT(0);
+      }
+      tqAddBlockDataToRsp(&block, pRsp);
+      if (pRsp->withTbName) {
+        tqAddTbNameToRsp(pTq, pExec, pRsp, workerId);
+      }
+      tqAddBlockSchemaToRsp(pExec, workerId, pRsp);
+      pRsp->blockNum++;
+    }
+  } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
+    pRsp->withSchema = 1;
+    STqReadHandle* pReader = pExec->pExecReader[workerId];
+    tqReadHandleSetMsg(pReader, pReq, 0);
+    while (tqNextDataBlockFilterOut(pReader, pExec->exec.execDb.pFilterOutTbUid)) {
+      SSDataBlock block = {0};
+      if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
+                              &block.info.numOfCols) < 0) {
+        if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;
+        ASSERT(0);
+      }
+      tqAddBlockDataToRsp(&block, pRsp);
+      if (pRsp->withTbName) {
+        tqAddTbNameToRsp(pTq, pExec, pRsp, workerId);
+      }
+      tqAddBlockSchemaToRsp(pExec, workerId, pRsp);
+      pRsp->blockNum++;
+    }
+  }
+  if (pRsp->blockNum == 0) {
+    pRsp->skipLogNum++;
+    return -1;
+  }
+  return 0;
+}

source/dnode/vnode/src/tq/tqMeta.c

+/*
+ * 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/>.
+ */

source/dnode/vnode/src/tq/tqRead.c

@@ -15,6 +15,48 @@
 
 #include "tq.h"
 
+int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalHead** ppHeadWithCkSum) {
+  int32_t code = 0;
+  taosThreadMutexLock(&pHandle->pWalReader->mutex);
+  int64_t offset = *fetchOffset;
+
+  while (1) {
+    if (walFetchHead(pHandle->pWalReader, offset, *ppHeadWithCkSum) < 0) {
+      tqDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", pHandle->consumerId,
+              pHandle->epoch, TD_VID(pTq->pVnode), offset);
+      *fetchOffset = offset - 1;
+      code = -1;
+      goto END;
+    }
+
+    if ((*ppHeadWithCkSum)->head.msgType == TDMT_VND_SUBMIT) {
+      code = walFetchBody(pHandle->pWalReader, ppHeadWithCkSum);
+
+      if (code < 0) {
+        ASSERT(0);
+        *fetchOffset = offset;
+        code = -1;
+        goto END;
+      }
+      *fetchOffset = offset;
+      code = 0;
+      goto END;
+    } else {
+      code = walSkipFetchBody(pHandle->pWalReader, *ppHeadWithCkSum);
+      if (code < 0) {
+        ASSERT(0);
+        *fetchOffset = offset;
+        code = -1;
+        goto END;
+      }
+      offset++;
+    }
+  }
+END:
+  taosThreadMutexUnlock(&pHandle->pWalReader->mutex);
+  return code;
+}
+
 STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
   STqReadHandle* pReadHandle = taosMemoryMalloc(sizeof(STqReadHandle));
   if (pReadHandle == NULL) {
@@ -24,7 +66,7 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
   pReadHandle->pMsg = NULL;
   pReadHandle->ver = -1;
   pReadHandle->pColIdList = NULL;
-  pReadHandle->sver = -1;
+  pReadHandle->cachedSchemaVer = -1;
   pReadHandle->cachedSchemaUid = -1;
   pReadHandle->pSchema = NULL;
   pReadHandle->pSchemaWrapper = NULL;
@@ -88,11 +130,11 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
   // TODO set to real sversion
   /*int32_t sversion = 1;*/
   int32_t sversion = htonl(pHandle->pBlock->sversion);
-  if (pHandle->sver != sversion || pHandle->cachedSchemaUid != pHandle->msgIter.suid) {
+  if (pHandle->cachedSchemaVer != sversion || pHandle->cachedSchemaUid != pHandle->msgIter.suid) {
     pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion);
     if (pHandle->pSchema == NULL) {
       tqWarn("cannot found tsschema for table: uid: %ld (suid: %ld), version %d, possibly dropped table",
-             pHandle->msgIter.uid, pHandle->msgIter.suid, pHandle->sver);
+             pHandle->msgIter.uid, pHandle->msgIter.suid, pHandle->cachedSchemaVer);
       /*ASSERT(0);*/
       terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
       return -1;
@@ -102,12 +144,12 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
     pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.suid, sversion, true);
     if (pHandle->pSchemaWrapper == NULL) {
       tqWarn("cannot found schema wrapper for table: suid: %ld, version %d, possibly dropped table",
-             pHandle->msgIter.suid, pHandle->sver);
+             pHandle->msgIter.suid, pHandle->cachedSchemaVer);
       /*ASSERT(0);*/
       terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
       return -1;
     }
-    pHandle->sver = sversion;
+    pHandle->cachedSchemaVer = sversion;
     pHandle->cachedSchemaUid = pHandle->msgIter.suid;
   }
 

tests/system-test/7-tmq/subscribeDb.py

@@ -182,7 +182,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
         topicList    = topicName1
@@ -223,7 +223,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
         topicList    = topicName1
@@ -279,7 +279,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
@@ -343,7 +343,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
@@ -427,7 +427,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]

tests/system-test/7-tmq/subscribeDb0.py

@@ -195,7 +195,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]
@@ -272,7 +272,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]
@@ -358,8 +358,8 @@ class TDTestCase:
         topicName1 = 'topic_db60'
         topicName2 = 'topic_db61'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))        
-        tdSql.execute("create topic %s as %s" %(topicName2, parameterDict2['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))        
+        tdSql.execute("create topic %s as database %s" %(topicName2, parameterDict2['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]
@@ -443,8 +443,8 @@ class TDTestCase:
         topicName1 = 'topic_db60'
         topicName2 = 'topic_db61'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))        
-        tdSql.execute("create topic %s as %s" %(topicName2, parameterDict2['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))        
+        tdSql.execute("create topic %s as database %s" %(topicName2, parameterDict2['dbName']))
         
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] +  parameterDict2["rowsPerTbl"] * parameterDict2["ctbNum"]

tests/system-test/7-tmq/subscribeDb1.py

@@ -183,7 +183,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] / 2
         topicList    = topicName1
@@ -261,7 +261,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"] / 2
         topicList    = topicName1
@@ -339,7 +339,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
         topicList    = topicName1
@@ -411,7 +411,7 @@ class TDTestCase:
         tdLog.info("create topics from db")
         topicName1 = 'topic_db1'
         
-        tdSql.execute("create topic %s as %s" %(topicName1, parameterDict['dbName']))
+        tdSql.execute("create topic %s as database %s" %(topicName1, parameterDict['dbName']))
         consumerId   = 0
         expectrowcnt = parameterDict["rowsPerTbl"] * parameterDict["ctbNum"]
         topicList    = topicName1