From b8f7526361bceda08972371088d995f0a7034964 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Fri, 13 May 2022 19:09:36 +0800
Subject: [PATCH] fix(tmq): msg schema

---
 source/dnode/vnode/src/inc/tq.h    | 1 +
 source/dnode/vnode/src/tq/tqRead.c | 8 +++++---
 2 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h
index 5eb89e8bb7..38dedee5a2 100644
--- a/source/dnode/vnode/src/inc/tq.h
+++ b/source/dnode/vnode/src/inc/tq.h
@@ -93,6 +93,7 @@ struct STqReadHandle {
   SMeta*            pVnodeMeta;
   SArray*           pColIdList;  // SArray<int16_t>
   int32_t           sver;
+  int64_t           cachedSchemaUid;
   SSchemaWrapper*   pSchemaWrapper;
   STSchema*         pSchema;
 };
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index f531d3f5fb..996d789e24 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -25,6 +25,7 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
   pReadHandle->ver = -1;
   pReadHandle->pColIdList = NULL;
   pReadHandle->sver = -1;
+  pReadHandle->cachedSchemaUid = -1;
   pReadHandle->pSchema = NULL;
   pReadHandle->pSchemaWrapper = NULL;
   pReadHandle->tbIdHash = NULL;
@@ -84,19 +85,20 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
   return false;
 }
 
-int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* pGroupId, uint64_t* pUid, int32_t* pNumOfRows,
-                            int16_t* pNumOfCols) {
+int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* pGroupId, uint64_t* pUid,
+                            int32_t* pNumOfRows, int16_t* pNumOfCols) {
   /*int32_t         sversion = pHandle->pBlock->sversion;*/
   // TODO set to real sversion
   *pUid = 0;
 
   int32_t sversion = 0;
-  if (pHandle->sver != sversion) {
+  if (pHandle->sver != sversion || pHandle->cachedSchemaUid != pHandle->msgIter.suid) {
     pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion);
 
     // this interface use suid instead of uid
     pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.suid, sversion, true);
     pHandle->sver = sversion;
+    pHandle->cachedSchemaUid = pHandle->msgIter.suid;
   }
 
   STSchema*       pTschema = pHandle->pSchema;