#5011

Committed 03 Apr 2026 03:59PM UTC coverage: 72.3% (+0.008%) from 72.292%

Build # #5011

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #35067

Coverage Stats

4053 of 5985 new or added lines in 68 files covered. (67.72%)

732 existing lines in 143 files now uncovered.

257430 of 356056 relevant lines covered (72.3%)

131834103.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.99

/source/dnode/vnode/src/tq/tqRead.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 "taoserror.h"
#include "tarray.h"
#include "tdef.h"
#include "thash.h"
#include "tmsg.h"
#include "tpriv.h"
#include "tq.h"

static int32_t tqRetrieveCols(STqReader *pReader, SSDataBlock *pRes, SHashObj* pCol2SlotId);

static void processCreateTbMsg(SDecoder* dcoder, SWalCont* pHead, STqReader* pReader, int64_t* realTbSuid, int64_t tbSuid) {
  int32_t code = 0;
  int32_t lino = 0;
  int32_t        needRebuild = 0;
  SVCreateTbReq* pCreateReq = NULL;
  SVCreateTbBatchReq reqNew = {0};
  void* buf = NULL;
  SVCreateTbBatchReq req = {0};
  code = tDecodeSVCreateTbBatchReq(dcoder, &req);
  if (code < 0) {
    lino = __LINE__;
    goto end;
  }

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
    pCreateReq = req.pReqs + iReq;
    if ((pCreateReq->type == TSDB_CHILD_TABLE || pCreateReq->type == TSDB_VIRTUAL_CHILD_TABLE) && 
         pCreateReq->ctb.suid == tbSuid &&
         taosHashGet(pReader->tbIdHash, &pCreateReq->uid, sizeof(int64_t)) != NULL) {  
      needRebuild++;
    }
  }
  if (needRebuild == 0) {
    // do nothing
  } else if (needRebuild == req.nReqs) {
    *realTbSuid = tbSuid;
  } else {
    *realTbSuid = tbSuid;
    reqNew.pArray = taosArrayInit(req.nReqs, sizeof(struct SVCreateTbReq));
    if (reqNew.pArray == NULL) {
      code = terrno;
      lino = __LINE__;
      goto end;
    }
    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
      pCreateReq = req.pReqs + iReq;
      if ((pCreateReq->type == TSDB_CHILD_TABLE || pCreateReq->type == TSDB_VIRTUAL_CHILD_TABLE) &&
          pCreateReq->ctb.suid == tbSuid &&
          taosHashGet(pReader->tbIdHash, &pCreateReq->uid, sizeof(int64_t)) != NULL) {
        reqNew.nReqs++;
        if (taosArrayPush(reqNew.pArray, pCreateReq) == NULL) {
          code = terrno;
          lino = __LINE__;
          goto end;
        }
      }
    }

    int     tlen = 0;
    tEncodeSize(tEncodeSVCreateTbBatchReq, &reqNew, tlen, code);
    buf = taosMemoryMalloc(tlen);
    if (NULL == buf || code < 0) {
      lino = __LINE__;
      goto end;
    }
    SEncoder coderNew = {0};
    tEncoderInit(&coderNew, buf, tlen);
    code = tEncodeSVCreateTbBatchReq(&coderNew, &reqNew);
    tEncoderClear(&coderNew);
    if (code < 0) {
      lino = __LINE__;
      goto end;
    }
    (void)memcpy(pHead->body + sizeof(SMsgHead), buf, tlen);
    pHead->bodyLen = tlen + sizeof(SMsgHead);
  }

end:
  taosMemoryFree(buf);
  taosArrayDestroy(reqNew.pArray);
  tDeleteSVCreateTbBatchReq(&req);
  if (code < 0) {
    tqError("processCreateTbMsg failed, code:%d, line:%d", code, lino);
  }
}

static void processAlterTbMsg(SDecoder* dcoder, SWalCont* pHead, STqReader* pReader, int64_t* realTbSuid, int64_t tbSuid) {
  SVAlterTbReq req = {0};
  SVAlterTbReq reqNew = {0};
  SMetaReader mr = {0};
  void* buf = NULL;
  int32_t lino = 0;
  int32_t code = tDecodeSVAlterTbReq(dcoder, &req);
  if (code < 0) {
    tqError("vgId:%d, processAlterTbMsg failed to decode SVAlterTbReq, code:%s",
            TD_VID(pReader->pVnode), tstrerror(code));
    lino = __LINE__;
    goto end;
  }

  if (req.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TABLE_TAG_VAL) {
    int32_t needRebuild = 0;
    for (int32_t i = 0; i < taosArrayGetSize(req.tables); i++) {
      SUpdateTableTagVal* pTable = taosArrayGet(req.tables, i);
      if (pTable == NULL || pTable->tbName == NULL) {
        tqWarn("vgId:%d, processAlterTbMsg Type 1 table[%d] has invalid name, skip",
               TD_VID(pReader->pVnode), i);
        continue;
      }

      metaReaderDoInit(&mr, pReader->pVnode->pMeta, META_READER_LOCK);
      code = metaGetTableEntryByName(&mr, pTable->tbName);
      if (code < 0) {
        // Table not found, skip
        tqDebug("vgId:%d, processAlterTbMsg Type 1 table %s not found, skip",
                TD_VID(pReader->pVnode), pTable->tbName);
        metaReaderClear(&mr);
        continue;
      }

      if (mr.me.ctbEntry.suid == tbSuid &&
          taosHashGet(pReader->tbIdHash, &mr.me.uid, sizeof(int64_t)) != NULL) {
        needRebuild++;
      }
      metaReaderClear(&mr);
    }

    if (needRebuild == 0) {
      // No tables in subscription scope, skip message
      tqDebug("vgId:%d, processAlterTbMsg Type 1: 0/%d tables in subscription, skip",
              TD_VID(pReader->pVnode), (int)taosArrayGetSize(req.tables));
    } else if (needRebuild == taosArrayGetSize(req.tables)) {
      // All tables in subscription scope, forward complete message
      *realTbSuid = tbSuid;
      tqDebug("vgId:%d, processAlterTbMsg Type 1: %d/%d tables in subscription, forward all",
              TD_VID(pReader->pVnode), needRebuild, (int)taosArrayGetSize(req.tables));
    } else {
      // Partial match: rebuild message with only subscribed tables
      *realTbSuid = tbSuid;
      tqDebug("vgId:%d, processAlterTbMsg Type 1: %d/%d tables in subscription, rebuild message",
              TD_VID(pReader->pVnode), needRebuild, (int)taosArrayGetSize(req.tables));

      // Build filtered message
      reqNew.action = req.action;
      reqNew.tbName = req.tbName;
      reqNew.tables = taosArrayInit(needRebuild, sizeof(SUpdateTableTagVal));
      if (reqNew.tables == NULL) {
        code = terrno;
        tqError("vgId:%d, processAlterTbMsg failed to allocate filtered tables array, code:%s",
                TD_VID(pReader->pVnode), tstrerror(code));
        lino = __LINE__;
        goto end;
      }

      // Collect only subscribed tables
      for (int32_t i = 0; i < taosArrayGetSize(req.tables); i++) {
        SUpdateTableTagVal* pTable = taosArrayGet(req.tables, i);
        if (pTable == NULL || pTable->tbName == NULL) {
          continue;
        }
        metaReaderDoInit(&mr, pReader->pVnode->pMeta, META_READER_LOCK);
        code = metaGetTableEntryByName(&mr, pTable->tbName);
        if (code < 0) {
          metaReaderClear(&mr);
          continue;
        }

        if (mr.me.ctbEntry.suid == tbSuid &&
            taosHashGet(pReader->tbIdHash, &mr.me.uid, sizeof(int64_t)) != NULL) {
          if (taosArrayPush(reqNew.tables, pTable) == NULL) {
            code = terrno;
            tqError("vgId:%d, processAlterTbMsg failed to add table %s to filtered array, code:%s",
                    TD_VID(pReader->pVnode), pTable->tbName, tstrerror(code));
            lino = __LINE__;
            metaReaderClear(&mr);
            goto end;
          }
        }
        metaReaderClear(&mr);
      }

      // Encode filtered message
      int tlen = 0;
      tEncodeSize(tEncodeSVAlterTbReq, &reqNew, tlen, code);
      if (code < 0) {
        tqError("vgId:%d, processAlterTbMsg failed to calculate encode size, code:%s",
                TD_VID(pReader->pVnode), tstrerror(code));
        lino = __LINE__;
        goto end;
      }

      buf = taosMemoryMalloc(tlen);
      if (NULL == buf) {
        code = terrno;
        tqError("vgId:%d, processAlterTbMsg failed to allocate encode buffer size:%d, code:%s",
                TD_VID(pReader->pVnode), tlen, tstrerror(code));
        lino = __LINE__;
        goto end;
      }

      SEncoder coderNew = {0};
      tEncoderInit(&coderNew, buf, tlen);
      code = tEncodeSVAlterTbReq(&coderNew, &reqNew);
      tEncoderClear(&coderNew);
      if (code != 0) {
        tqError("vgId:%d, processAlterTbMsg failed to encode filtered message, code:%s",
                TD_VID(pReader->pVnode), tstrerror(code));
        lino = __LINE__;
        goto end;
      }
      (void)memcpy(pHead->body + sizeof(SMsgHead), buf, tlen);
      pHead->bodyLen = tlen + sizeof(SMsgHead);
      tqInfo("vgId:%d, processAlterTbMsg Type 1 rebuilt message with %d/%d tables",
             TD_VID(pReader->pVnode), needRebuild, (int)taosArrayGetSize(req.tables));
    }

  } else if (req.action == TSDB_ALTER_TABLE_UPDATE_CHILD_TABLE_TAG_VAL) {
    // Type 2: WHERE condition on super table
    metaReaderDoInit(&mr, pReader->pVnode->pMeta, META_READER_LOCK);
    code = metaGetTableEntryByName(&mr, req.tbName);
    if (code < 0) {
      tqError("vgId:%d, processAlterTbMsg Type 2 failed to get super table %s, code:%s",
              TD_VID(pReader->pVnode), req.tbName, tstrerror(code));
      lino = __LINE__;
      metaReaderClear(&mr);
      goto end;
    }

    // For Type 2, the super table is specified in req.tbName
    // The actual filtering happens through tbIdHash which already contains
    // only the child tables matching the topic's WHERE condition
    if (mr.me.type == TSDB_SUPER_TABLE) {
      *realTbSuid = mr.me.uid;
      tqDebug("vgId:%d, processAlterTbMsg Type 2: super table %" PRId64 " (WHERE condition filtering via tbIdHash)",
              TD_VID(pReader->pVnode), mr.me.uid);
    }
    metaReaderClear(&mr);
  } else {
    // Legacy single-table tag modification
    metaReaderDoInit(&mr, pReader->pVnode->pMeta, META_READER_LOCK);
    code = metaGetTableEntryByName(&mr, req.tbName);
    if (code < 0) {
      tqDebug("vgId:%d, processAlterTbMsg legacy type failed to get table %s, code:%s",
              TD_VID(pReader->pVnode), req.tbName ? req.tbName : "(null)", tstrerror(code));
      lino = __LINE__;
      metaReaderClear(&mr);
      goto end;
    }
    if (taosHashGet(pReader->tbIdHash, &mr.me.uid, sizeof(int64_t)) != NULL) {
      *realTbSuid = mr.me.ctbEntry.suid;
    }
    metaReaderClear(&mr);
  }

end:
  taosMemoryFree(buf);
  taosArrayDestroy(reqNew.tables);
  destroyAlterTbReq(&req);
  if (code < 0) {
    tqError("vgId:%d, processAlterTbMsg failed at line:%d, code:%s",
            TD_VID(pReader->pVnode), lino, tstrerror(code));
  }
} 

static void processDropTbMsg(SDecoder* dcoder, SWalCont* pHead, STqReader* pReader, int64_t* realTbSuid, int64_t tbSuid) {
  SVDropTbBatchReq req = {0};
  SVDropTbBatchReq reqNew = {0};
  void* buf = NULL;
  int32_t lino = 0;
  int32_t code = tDecodeSVDropTbBatchReq(dcoder, &req);
  if (code < 0) {
    lino = __LINE__;
    goto end;
  }

  int32_t      needRebuild = 0;
  SVDropTbReq* pDropReq = NULL;
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
    pDropReq = req.pReqs + iReq;

    if (pDropReq->suid == tbSuid &&
        taosHashGet(pReader->tbIdHash, &pDropReq->uid, sizeof(int64_t)) != NULL) {
      needRebuild++;
    }
  }
  if (needRebuild == 0) {
    // do nothing
  } else if (needRebuild == req.nReqs) {
    *realTbSuid = tbSuid;
  } else {
    *realTbSuid = tbSuid;
    reqNew.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbReq));
    if (reqNew.pArray == NULL) {
      code = terrno;
      lino = __LINE__;
      goto end;
    }
    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
      pDropReq = req.pReqs + iReq;
      if (pDropReq->suid == tbSuid &&
          taosHashGet(pReader->tbIdHash, &pDropReq->uid, sizeof(int64_t)) != NULL) {
        reqNew.nReqs++;
        if (taosArrayPush(reqNew.pArray, pDropReq) == NULL) {
          code = terrno;
          lino = __LINE__;
          goto end;
        }
      }
    }

    int     tlen = 0;
    tEncodeSize(tEncodeSVDropTbBatchReq, &reqNew, tlen, code);
    buf = taosMemoryMalloc(tlen);
    if (NULL == buf || code < 0) {
      lino = __LINE__;
      goto end;
    }
    SEncoder coderNew = {0};
    tEncoderInit(&coderNew, buf, tlen);
    code = tEncodeSVDropTbBatchReq(&coderNew, &reqNew);
    tEncoderClear(&coderNew);
    if (code != 0) {
      lino = __LINE__;
      goto end;
    }
    (void)memcpy(pHead->body + sizeof(SMsgHead), buf, tlen);
    pHead->bodyLen = tlen + sizeof(SMsgHead);
  }

end:
  taosMemoryFree(buf);
  taosArrayDestroy(reqNew.pArray);
  if (code < 0) {
    tqError("processDropTbMsg failed, code:%d, line:%d", code, lino);
  }
}

bool isValValidForTable(STqHandle* pHandle, SWalCont* pHead) {
  int32_t code = 0;
  int32_t lino = 0;
  if (pHandle == NULL || pHead == NULL) {
    return false;
  }
  if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__TABLE) {
    return true;
  }

  STqExecHandle* pExec = &pHandle->execHandle;
  STqReader* pReader = pExec->pTqReader;

  int16_t msgType = pHead->msgType;
  char*   body = pHead->body;
  int32_t bodyLen = pHead->bodyLen;

  int64_t  tbSuid = pHandle->execHandle.execTb.suid;
  int64_t  realTbSuid = 0;
  SDecoder dcoder = {0};
  void*    data = POINTER_SHIFT(body, sizeof(SMsgHead));
  int32_t  len = bodyLen - sizeof(SMsgHead);
  tDecoderInit(&dcoder, data, len);

  if (msgType == TDMT_VND_CREATE_STB || msgType == TDMT_VND_ALTER_STB) {
    SVCreateStbReq req = {0};
    if (tDecodeSVCreateStbReq(&dcoder, &req) < 0) {
      goto end;
    }
    realTbSuid = req.suid;
  } else if (msgType == TDMT_VND_DROP_STB) {
    SVDropStbReq req = {0};
    if (tDecodeSVDropStbReq(&dcoder, &req) < 0) {
      goto end;
    }
    realTbSuid = req.suid;
  } else if (msgType == TDMT_VND_CREATE_TABLE) {
    processCreateTbMsg(&dcoder, pHead, pReader, &realTbSuid, tbSuid);
  } else if (msgType == TDMT_VND_ALTER_TABLE) {
    processAlterTbMsg(&dcoder, pHead, pReader, &realTbSuid, tbSuid);
  } else if (msgType == TDMT_VND_DROP_TABLE) {
    processDropTbMsg(&dcoder, pHead, pReader, &realTbSuid, tbSuid);
  } else if (msgType == TDMT_VND_DELETE) {
    SDeleteRes req = {0};
    if (tDecodeDeleteRes(&dcoder, &req) < 0) {
      goto end;
    }
    realTbSuid = req.suid;
  }

end:
  tDecoderClear(&dcoder);
  bool tmp = tbSuid == realTbSuid;
  tqDebug("%s suid:%" PRId64 " realSuid:%" PRId64 " return:%d", __FUNCTION__, tbSuid, realTbSuid, tmp);
  return tmp;
}

int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, uint64_t reqId) {
  if (pTq == NULL || pHandle == NULL || fetchOffset == NULL) {
    return -1;
  }
  int32_t code = -1;
  int32_t vgId = TD_VID(pTq->pVnode);
  int64_t id = pHandle->pWalReader->readerId;

  int64_t offset = *fetchOffset;
  int64_t lastVer = walGetLastVer(pHandle->pWalReader->pWal);
  int64_t committedVer = walGetCommittedVer(pHandle->pWalReader->pWal);
  int64_t appliedVer = walGetAppliedVer(pHandle->pWalReader->pWal);

  tqDebug("vgId:%d, start to fetch wal, index:%" PRId64 ", last:%" PRId64 " commit:%" PRId64 ", applied:%" PRId64
          ", 0x%" PRIx64,
          vgId, offset, lastVer, committedVer, appliedVer, id);

  while (offset <= appliedVer) {
    if (walFetchHead(pHandle->pWalReader, offset) < 0) {
      tqDebug("tmq poll: consumer:0x%" PRIx64 ", (epoch %d) vgId:%d offset %" PRId64
              ", no more log to return, QID:0x%" PRIx64 " 0x%" PRIx64,
              pHandle->consumerId, pHandle->epoch, vgId, offset, reqId, id);
      goto END;
    }

    tqDebug("vgId:%d, consumer:0x%" PRIx64 " taosx get msg ver %" PRId64 ", type:%s, QID:0x%" PRIx64 " 0x%" PRIx64,
            vgId, pHandle->consumerId, offset, TMSG_INFO(pHandle->pWalReader->pHead->head.msgType), reqId, id);

    if (pHandle->pWalReader->pHead->head.msgType == TDMT_VND_SUBMIT) {
      code = walFetchBody(pHandle->pWalReader);
      goto END;
    } else {
      if (pHandle->fetchMeta != WITH_DATA) {
        SWalCont* pHead = &(pHandle->pWalReader->pHead->head);
        if (IS_META_MSG(pHead->msgType) && !(pHead->msgType == TDMT_VND_DELETE && pHandle->fetchMeta == ONLY_META)) {
          code = walFetchBody(pHandle->pWalReader);
          if (code < 0) {
            goto END;
          }

          pHead = &(pHandle->pWalReader->pHead->head);
          if (isValValidForTable(pHandle, pHead)) {
            code = 0;
            goto END;
          } else {
            offset++;
            code = -1;
            continue;
          }
        }
      }
      code = walSkipFetchBody(pHandle->pWalReader);
      if (code < 0) {
        goto END;
      }
      offset++;
    }
    code = -1;
  }

END:
  *fetchOffset = offset;
  tqDebug("vgId:%d, end to fetch wal, code:%d , index:%" PRId64 ", last:%" PRId64 " commit:%" PRId64
          ", applied:%" PRId64 ", 0x%" PRIx64,
          vgId, code, offset, lastVer, committedVer, appliedVer, id);
  return code;
}

bool tqGetTablePrimaryKey(STqReader* pReader) {
  if (pReader == NULL) {
    return false;
  }
  return pReader->hasPrimaryKey;
}

void tqSetTablePrimaryKey(STqReader* pReader, int64_t uid) {
  tqDebug("%s:%p uid:%" PRId64, __FUNCTION__, pReader, uid);

  if (pReader == NULL) {
    return;
  }
  bool            ret = false;
  SSchemaWrapper* schema = metaGetTableSchema(pReader->pVnode->pMeta, uid, -1, 1, NULL, 0);
  if (schema && schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY) {
    ret = true;
  }
  tDeleteSchemaWrapper(schema);
  pReader->hasPrimaryKey = ret;
}

static void freeTagCache(void* pData){
  if (pData == NULL) return;
  SArray* tagCache = *(SArray**)pData;
  taosArrayDestroyP(tagCache, taosMemFree);
}

STqReader* tqReaderOpen(SVnode* pVnode) {
  tqDebug("%s:%p", __FUNCTION__, pVnode);
  if (pVnode == NULL) {
    return NULL;
  }
  STqReader* pReader = taosMemoryCalloc(1, sizeof(STqReader));
  if (pReader == NULL) {
    return NULL;
  }

  pReader->pWalReader = walOpenReader(pVnode->pWal, 0);
  if (pReader->pWalReader == NULL) {
    taosMemoryFree(pReader);
    return NULL;
  }

  pReader->pVnode = pVnode;
  pReader->pSchemaWrapper = NULL;
  pReader->tbIdHash = NULL;
  pReader->pTableTagCacheForTmq = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
  if (pReader->pTableTagCacheForTmq == NULL) {
    walCloseReader(pReader->pWalReader);
    taosMemoryFree(pReader);
    return NULL;
  }
  taosHashSetFreeFp(pReader->pTableTagCacheForTmq, freeTagCache);
  taosInitRWLatch(&pReader->tagCachelock);

  return pReader;
}

void tqReaderClose(STqReader* pReader) {
  tqDebug("%s:%p", __FUNCTION__, pReader);
  if (pReader == NULL) return;

  // close wal reader
  walCloseReader(pReader->pWalReader);
  taosHashCleanup(pReader->pTableTagCacheForTmq);
  tDeleteSchemaWrapper(pReader->pSchemaWrapper);
  taosMemoryFree(pReader->pTSchema);
  taosMemoryFree(pReader->extSchema);

  // free hash
  taosHashCleanup(pReader->tbIdHash);
  tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);

  taosMemoryFree(pReader);
}

int32_t tqReaderSeek(STqReader* pReader, int64_t ver, const char* id) {
  if (pReader == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  if (walReaderSeekVer(pReader->pWalReader, ver) < 0) {
    return terrno;
  }
  tqDebug("wal reader seek to ver:%" PRId64 " %s", ver, id);
  return 0;
}

static int32_t getTableTagCache(STqReader* pReader, SExprInfo* pExprInfo, int32_t numOfExpr, int64_t uid) {
  int32_t code = 0;
  int32_t lino = 0;

  void* data = taosHashGet(pReader->pTableTagCacheForTmq, &uid, LONG_BYTES);
  if (data == NULL) {
    SStorageAPI api = {0}; 
    initStorageAPI(&api);
    code = cacheTag(pReader->pVnode, pReader->pTableTagCacheForTmq, pExprInfo, numOfExpr, &api, uid, 0, &pReader->tagCachelock);
    TSDB_CHECK_CODE(code, lino, END);
  }

  END:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);
  }
  
  return code;
}

void tqUpdateTableTagCache(STqReader* pReader, SExprInfo* pExprInfo, int32_t numOfExpr, int64_t uid, col_id_t colId) {
  int32_t code = 0;
  int32_t lino = 0;

  void* data = taosHashGet(pReader->pTableTagCacheForTmq, &uid, LONG_BYTES);
  if (data == NULL) {
    return;
  }

  SStorageAPI api = {0}; 
  initStorageAPI(&api);
  code = cacheTag(pReader->pVnode, pReader->pTableTagCacheForTmq, pExprInfo, numOfExpr, &api, uid, colId, &pReader->tagCachelock);
  TSDB_CHECK_CODE(code, lino, END);

  END:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at %d, failed to update tag cache code:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);
  }
}

static int32_t tqRetrievePseudoCols(STqReader* pReader, SSDataBlock* pBlock, int32_t numOfRows, int64_t uid, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr) {
  if (pReader == NULL || pBlock == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  
  code = getTableTagCache(pReader, pPseudoExpr, numOfPseudoExpr, uid);
  TSDB_CHECK_CODE(code, lino, END);

  code = fillTag(pReader->pTableTagCacheForTmq, pPseudoExpr, numOfPseudoExpr, uid, pBlock, numOfRows, pBlock->info.rows - numOfRows, 1, &pReader->tagCachelock);
  TSDB_CHECK_CODE(code, lino, END);

END:
  if (code != 0) {
    tqError("tqRetrievePseudoCols failed, line:%d, msg:%s", lino, tstrerror(code));
  }
  return code;
}

int32_t tqNextBlockInWal(STqReader* pReader, SSDataBlock* pRes, SHashObj* pCol2SlotId, SExprInfo* pPseudoExpr, int32_t numOfPseudoExpr,
                         int sourceExcluded, int32_t minPollRows, int64_t timeout, int8_t enableReplay) {
  int32_t code = 0;
  if (pReader == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  SWalReader* pWalReader = pReader->pWalReader;

  int64_t st = taosGetTimestampMs();
  while (1) {
    code = walNextValidMsg(pWalReader, false);
    if (code != 0) {
      break;
    }

    void*   pBody = POINTER_SHIFT(pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg));
    int32_t bodyLen = pWalReader->pHead->head.bodyLen - sizeof(SSubmitReq2Msg);
    int64_t ver = pWalReader->pHead->head.version;
    SDecoder decoder = {0};
    code = tqReaderSetSubmitMsg(pReader, pBody, bodyLen, ver, NULL, &decoder);
    tDecoderClear(&decoder);
    if (code != 0) {
      return code;
    }
    pReader->nextBlk = 0;

    int32_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData);
    while (pReader->nextBlk < numOfBlocks) {
      tqDebug("tq reader next data block %d/%d, len:%d %" PRId64, pReader->nextBlk, numOfBlocks, pReader->msg.msgLen,
              pReader->msg.ver);

      SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
      if (pSubmitTbData == NULL) {
        tqError("tq reader next data block %d/%d, len:%d %" PRId64, pReader->nextBlk, numOfBlocks, pReader->msg.msgLen,
                pReader->msg.ver);
        return terrno;
      }
      if ((pSubmitTbData->flags & sourceExcluded) != 0) {
        pReader->nextBlk += 1;
        continue;
      }
      if (pReader->tbIdHash == NULL || taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t)) != NULL) {
        tqDebug("tq reader return submit block, uid:%" PRId64, pSubmitTbData->uid);
        int32_t numOfRows = pRes->info.rows;
        code = tqRetrieveCols(pReader, pRes, pCol2SlotId);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
        code = tqRetrievePseudoCols(pReader, pRes, numOfRows, pSubmitTbData->uid, pPseudoExpr, numOfPseudoExpr);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }

      }
      pReader->nextBlk += 1;
    }

    tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
    pReader->msg.msgStr = NULL;

    if (pRes->info.rows >= minPollRows || (enableReplay && pRes->info.rows > 0)){
      break;
    }
    int64_t elapsed = taosGetTimestampMs() - st;
    if (elapsed > timeout || elapsed < 0) {
      code = TSDB_CODE_TMQ_FETCH_TIMEOUT;
      terrno = code;
      break;
    }
  }
  return code;
}

int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, int64_t ver, SArray* rawList, SDecoder* decoder) {
  if (pReader == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  pReader->msg.msgStr = msgStr;
  pReader->msg.msgLen = msgLen;
  pReader->msg.ver = ver;

  tqTrace("tq reader set msg pointer:%p, msg len:%d", msgStr, msgLen);

  tDecoderInit(decoder, pReader->msg.msgStr, pReader->msg.msgLen);
  int32_t code = tDecodeSubmitReq(decoder, &pReader->submit, rawList);

  if (code != 0) {
    tqError("DecodeSSubmitReq2 error, msgLen:%d, ver:%" PRId64, msgLen, ver);
  }

  return code;
}

void tqReaderClearSubmitMsg(STqReader* pReader) {
  tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
  pReader->nextBlk = 0;
  pReader->msg.msgStr = NULL;
}

SWalReader* tqGetWalReader(STqReader* pReader) {
  if (pReader == NULL) {
    return NULL;
  }
  return pReader->pWalReader;
}

int64_t tqGetResultBlockTime(STqReader* pReader) {
  if (pReader == NULL) {
    return 0;
  }
  return pReader->lastTs;
}

bool tqNextBlockImpl(STqReader* pReader, const char* idstr) {
  int32_t code = false;
  int32_t lino = 0;
  int64_t uid = 0;
  TSDB_CHECK_NULL(pReader, code, lino, END, false);
  TSDB_CHECK_NULL(pReader->msg.msgStr, code, lino, END, false);
  TSDB_CHECK_NULL(pReader->tbIdHash, code, lino, END, true);

  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
  while (pReader->nextBlk < blockSz) {
    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
    TSDB_CHECK_NULL(pSubmitTbData, code, lino, END, false);
    uid = pSubmitTbData->uid;
    void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t));
    TSDB_CHECK_CONDITION(ret == NULL, code, lino, END, true);

    tqTrace("iterator data block in hash continue, progress:%d/%d, total queried tables:%d, uid:%" PRId64,
            pReader->nextBlk, blockSz, taosHashGetSize(pReader->tbIdHash), uid);
    pReader->nextBlk++;
  }

  tqReaderClearSubmitMsg(pReader);
  tqTrace("iterator data block end, total block num:%d, uid:%" PRId64, blockSz, uid);

END:
  tqTrace("%s:%d return:%s, uid:%" PRId64, __FUNCTION__, lino, code ? "true" : "false", uid);
  return code;
}

bool tqNextDataBlockFilterOut(STqReader* pReader, SHashObj* filterOutUids) {
  int32_t code = false;
  int32_t lino = 0;
  int64_t uid = 0;

  TSDB_CHECK_NULL(pReader, code, lino, END, false);
  TSDB_CHECK_NULL(pReader->msg.msgStr, code, lino, END, false);
  TSDB_CHECK_NULL(filterOutUids, code, lino, END, true);

  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
  while (pReader->nextBlk < blockSz) {
    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
    TSDB_CHECK_NULL(pSubmitTbData, code, lino, END, false);
    uid = pSubmitTbData->uid;
    void* ret = taosHashGet(filterOutUids, &pSubmitTbData->uid, sizeof(int64_t));
    TSDB_CHECK_NULL(ret, code, lino, END, true);
    tqTrace("iterator data block in hash jump block, progress:%d/%d, uid:%" PRId64, pReader->nextBlk, blockSz, uid);
    pReader->nextBlk++;
  }
  tqReaderClearSubmitMsg(pReader);
  tqTrace("iterator data block end, total block num:%d, uid:%" PRId64, blockSz, uid);

END:
  tqTrace("%s:%d get data:%s, uid:%" PRId64, __FUNCTION__, lino, code ? "true" : "false", uid);
  return code;
}

int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrapper* pSrc, char* mask,
                    SExtSchema* extSrc) {
  if (pDst == NULL || pBlock == NULL || pSrc == NULL || mask == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  int32_t code = 0;

  int32_t cnt = 0;
  for (int32_t i = 0; i < pSrc->nCols; i++) {
    cnt += mask[i];
  }

  pDst->nCols = cnt;
  pDst->pSchema = taosMemoryCalloc(cnt, sizeof(SSchema));
  if (pDst->pSchema == NULL) {
    return TAOS_GET_TERRNO(terrno);
  }

  int32_t j = 0;
  for (int32_t i = 0; i < pSrc->nCols; i++) {
    if (mask[i]) {
      pDst->pSchema[j++] = pSrc->pSchema[i];
      SColumnInfoData colInfo =
          createColumnInfoData(pSrc->pSchema[i].type, pSrc->pSchema[i].bytes, pSrc->pSchema[i].colId);
      if (extSrc != NULL) {
        decimalFromTypeMod(extSrc[i].typeMod, &colInfo.info.precision, &colInfo.info.scale);
      }
      code = blockDataAppendColInfo(pBlock, &colInfo);
      if (code != 0) {
        return code;
      }
    }
  }
  return 0;
}

static int32_t doSetBlobVal(SColumnInfoData* pColumnInfoData, int32_t idx, SColVal* pColVal, SBlobSet* pBlobRow2) {
  int32_t code = 0;
  if (pColumnInfoData == NULL || pColVal == NULL || pBlobRow2 == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  // TODO(yhDeng)
  if (COL_VAL_IS_VALUE(pColVal)) {
    char* val = taosMemCalloc(1, pColVal->value.nData + sizeof(BlobDataLenT));
    if (val == NULL) {
      return terrno;
    }

    uint64_t seq = 0;
    int32_t  len = 0;
    if (pColVal->value.pData != NULL) {
      if (tGetU64(pColVal->value.pData, &seq) < 0){
        TAOS_CHECK_RETURN(TSDB_CODE_INVALID_PARA);
      }
      SBlobItem item = {0};
      code = tBlobSetGet(pBlobRow2, seq, &item);
      if (code != 0) {
        taosMemoryFree(val);
        terrno = code;
        uError("tq set blob val, idx:%d, get blob item failed, seq:%" PRIu64 ", code:%d", idx, seq, code);
        return code;
      }

      val = taosMemRealloc(val, item.len + sizeof(BlobDataLenT));
      (void)memcpy(blobDataVal(val), item.data, item.len);
      len = item.len;
    }

    blobDataSetLen(val, len);
    code = colDataSetVal(pColumnInfoData, idx, val, false);

    taosMemoryFree(val);
  } else {
    colDataSetNULL(pColumnInfoData, idx);
  }
  return code;
}
static int32_t doSetVal(SColumnInfoData* pColumnInfoData, int32_t rowIndex, SColVal* pColVal) {
  int32_t code = TSDB_CODE_SUCCESS;

  if (IS_VAR_DATA_TYPE(pColVal->value.type)) {
    if (COL_VAL_IS_VALUE(pColVal)) {
      char val[65535 + 2] = {0};
      if (pColVal->value.pData != NULL) {
        (void)memcpy(varDataVal(val), pColVal->value.pData, pColVal->value.nData);
      }
      varDataSetLen(val, pColVal->value.nData);
      code = colDataSetVal(pColumnInfoData, rowIndex, val, false);
    } else {
      colDataSetNULL(pColumnInfoData, rowIndex);
    }
  } else {
    code = colDataSetVal(pColumnInfoData, rowIndex, VALUE_GET_DATUM(&pColVal->value, pColVal->value.type),
                         !COL_VAL_IS_VALUE(pColVal));
  }

  return code;
}

static int32_t setBlockData(SSDataBlock* pBlock, int32_t slotId, int32_t rowIndex, SColVal* colVal, SBlobSet* pBlobSet) {
  int32_t        code = 0;
  SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, slotId);
  if (pColData == NULL) {
    return terrno;
  }

  uint8_t isBlob = IS_STR_DATA_BLOB(pColData->info.type) ? 1 : 0;
  if (isBlob == 0) {
    code = doSetVal(pColData, rowIndex, colVal);
  } else {
    code = doSetBlobVal(pColData, rowIndex, colVal, pBlobSet);
  }
  return code;
}

static int32_t processSubmitRow(SArray*         pRows,
                                SSDataBlock*    pBlock,
                                SHashObj*       pCol2SlotId,
                                STqReader*      pReader,
                                SBlobSet*       pBlobSet) {
  int32_t        code = 0;
  int32_t        line = 0;

  SArray* pColArray = taosArrayInit(4, INT_BYTES * 2);
  TSDB_CHECK_NULL(pColArray, code, line, END, terrno);

  int32_t sourceIdx = -1;
  int32_t rowIndex = 0;
  SRow* pRow = taosArrayGetP(pRows, rowIndex);
  TSDB_CHECK_NULL(pRow, code, line, END, terrno);
  while (++sourceIdx < pReader->pTSchema->numOfCols) {
    SColVal colVal = {0};
    code = tRowGet(pRow, pReader->pTSchema, sourceIdx, &colVal);
    TSDB_CHECK_CODE(code, line, END);
    void* pSlotId = taosHashGet(pCol2SlotId, &colVal.cid, sizeof(colVal.cid));
    if (pSlotId == NULL) {
      continue;
    }
    int32_t pData[2] = {sourceIdx, *(int16_t*)pSlotId};
    TSDB_CHECK_NULL(taosArrayPush(pColArray, pData), code, line, END, terrno);
    code = setBlockData(pBlock, pData[1], pBlock->info.rows + rowIndex, &colVal, pBlobSet);
    TSDB_CHECK_CODE(code, line, END);
  }
  
  for (rowIndex = 1; rowIndex < taosArrayGetSize(pRows); rowIndex++) {
    SRow* pRow = taosArrayGetP(pRows, rowIndex);
    TSDB_CHECK_NULL(pRow, code, line, END, terrno);
    for (int32_t j = 0; j < taosArrayGetSize(pColArray); j++) {
      int32_t* pData = taosArrayGet(pColArray, j);
      TSDB_CHECK_NULL(pData, code, line, END, terrno);

      SColVal colVal = {0};
      code = tRowGet(pRow, pReader->pTSchema, pData[0], &colVal);
      TSDB_CHECK_CODE(code, line, END);

      code = setBlockData(pBlock, pData[1], pBlock->info.rows + rowIndex, &colVal, pBlobSet);
      TSDB_CHECK_CODE(code, line, END);
    }
  }

  END:
  taosArrayDestroy(pColArray);
  return code;
}

static int32_t processSubmitCol(SArray*         pCols,
                                SSDataBlock*    pBlock,
                                SHashObj*       pCol2SlotId,
                                SBlobSet*       pBlobSet) {
  int32_t        code = 0;
  int32_t        line = 0;

  for (int32_t i = 0; i < taosArrayGetSize(pCols); i++) {
    SColData* pCol = taosArrayGet(pCols, i);
    TSDB_CHECK_NULL(pCol, code, line, END, terrno);
    void* pSlotId = taosHashGet(pCol2SlotId, &pCol->cid, sizeof(pCol->cid));
    if (pSlotId == NULL) {
      continue;
    }
    SColVal colVal = {0};
    for (int32_t row = 0; row < pCol->nVal; row++) {
      code = tColDataGetValue(pCol, row, &colVal);
      TSDB_CHECK_CODE(code, line, END);

      code = setBlockData(pBlock, *(int16_t*)pSlotId, pBlock->info.rows + row, &colVal, pBlobSet);
      TSDB_CHECK_CODE(code, line, END);
    }
  }
  
  END:
  return code;
}

static int32_t checkSchema(STqReader* pReader, SSubmitTbData* pSubmitTbData) {
  int32_t vgId = pReader->pWalReader->pWal->cfg.vgId;
  int32_t sversion = pSubmitTbData->sver;
  int64_t suid = pSubmitTbData->suid;
  int64_t uid = pSubmitTbData->uid;
  if ((suid != 0 && pReader->cachedSchemaSuid != suid) || (suid == 0 && pReader->cachedSchemaUid != uid) ||
      (pReader->cachedSchemaVer != sversion)) {
    tDeleteSchemaWrapper(pReader->pSchemaWrapper);
    taosMemoryFreeClear(pReader->extSchema);
    taosMemoryFreeClear(pReader->pTSchema);
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnode->pMeta, uid, sversion, 1, &pReader->extSchema, 0);
    if (pReader->pSchemaWrapper == NULL) {
      tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", uid:%" PRId64 ",version %d, possibly dropped table",
              vgId, suid, uid, pReader->cachedSchemaVer);
      return TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
    }
    pReader->pTSchema = tBuildTSchema(pReader->pSchemaWrapper->pSchema, pReader->pSchemaWrapper->nCols, pReader->pSchemaWrapper->version);
    if (pReader->pTSchema == NULL) {
      tqWarn("vgId:%d, cannot build schema for table: suid:%" PRId64 ", uid:%" PRId64 ",version %d",
              vgId, suid, uid, pReader->cachedSchemaVer);
      return terrno;
    }
    pReader->cachedSchemaUid = uid;
    pReader->cachedSchemaSuid = suid;
    pReader->cachedSchemaVer = sversion;
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t tqRetrieveCols(STqReader* pReader, SSDataBlock* pBlock, SHashObj* pCol2SlotId) {
  if (pReader == NULL || pBlock == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  tqDebug("tq reader retrieve data block %p, index:%d", pReader->msg.msgStr, pReader->nextBlk);
  int32_t        code = 0;
  int32_t        line = 0;
  SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
  TSDB_CHECK_NULL(pSubmitTbData, code, line, END, terrno);
  pReader->lastTs = pSubmitTbData->ctimeMs;

  int32_t numOfRows = 0;
  if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
    SColData* pCol = taosArrayGet(pSubmitTbData->aCol, 0);
    TSDB_CHECK_NULL(pCol, code, line, END, terrno);
    numOfRows = pCol->nVal;
  } else {
    numOfRows = taosArrayGetSize(pSubmitTbData->aRowP);
  }

  code = blockDataEnsureCapacity(pBlock, pBlock->info.rows + numOfRows);
  TSDB_CHECK_CODE(code, line, END);

  code = checkSchema(pReader, pSubmitTbData);
  TSDB_CHECK_CODE(code, line, END);

  // convert and scan one block
  if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
    SArray* pCols = pSubmitTbData->aCol;
    code = processSubmitCol(pCols, pBlock, pCol2SlotId, pSubmitTbData->pBlobSet);
    TSDB_CHECK_CODE(code, line, END);
  } else {
    SArray*         pRows = pSubmitTbData->aRowP;
    code = processSubmitRow(pRows, pBlock, pCol2SlotId, pReader, pSubmitTbData->pBlobSet);
    TSDB_CHECK_CODE(code, line, END);
  }
  pBlock->info.rows += numOfRows;
END:
  if (code != 0) {
    tqError("tqRetrieveCols failed, line:%d, msg:%s", line, tstrerror(code));
  }
  return code;
}

#define PROCESS_VAL                                      \
  if (curRow == 0) {                                     \
    assigned[j] = !COL_VAL_IS_NONE(&colVal);             \
    buildNew = true;                                     \
  } else {                                               \
    bool currentRowAssigned = !COL_VAL_IS_NONE(&colVal); \
    if (currentRowAssigned != assigned[j]) {             \
      assigned[j] = currentRowAssigned;                  \
      buildNew = true;                                   \
    }                                                    \
  }

#define SET_DATA                                                                                    \
  if (colVal.cid < pColData->info.colId) {                                                          \
    sourceIdx++;                                                                                    \
  } else if (colVal.cid == pColData->info.colId) {                                                  \
    if (IS_STR_DATA_BLOB(pColData->info.type)) {                                                    \
      TQ_ERR_GO_TO_END(doSetBlobVal(pColData, curRow - lastRow, &colVal, pSubmitTbData->pBlobSet)); \
    } else {                                                                                        \
      TQ_ERR_GO_TO_END(doSetVal(pColData, curRow - lastRow, &colVal));                              \
    }                                                                                               \
    sourceIdx++;                                                                                    \
    targetIdx++;                                                                                    \
  } else {                                                                                          \
    colDataSetNULL(pColData, curRow - lastRow);                                                     \
    targetIdx++;                                                                                    \
  }

static int32_t processBuildNew(STqReader* pReader, SSubmitTbData* pSubmitTbData, SArray* blocks, SArray* schemas,
                               char* assigned, int32_t numOfRows, int32_t curRow, int32_t* lastRow) {
  int32_t         code = 0;
  SSchemaWrapper* pSW = NULL;
  SSDataBlock*    block = NULL;
  if (taosArrayGetSize(blocks) > 0) {
    SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
    TQ_NULL_GO_TO_END(pLastBlock);
    pLastBlock->info.rows = curRow - *lastRow;
    *lastRow = curRow;
  }

  block = taosMemoryCalloc(1, sizeof(SSDataBlock));
  TQ_NULL_GO_TO_END(block);

  pSW = taosMemoryCalloc(1, sizeof(SSchemaWrapper));
  TQ_NULL_GO_TO_END(pSW);

  TQ_ERR_GO_TO_END(tqMaskBlock(pSW, block, pReader->pSchemaWrapper, assigned, pReader->extSchema));
  tqTrace("vgId:%d, build new block, col %d", pReader->pWalReader->pWal->cfg.vgId,
          (int32_t)taosArrayGetSize(block->pDataBlock));

  block->info.id.uid = pSubmitTbData->uid;
  block->info.version = pReader->msg.ver;
  TQ_ERR_GO_TO_END(blockDataEnsureCapacity(block, numOfRows - curRow));
  TQ_NULL_GO_TO_END(taosArrayPush(blocks, block));
  TQ_NULL_GO_TO_END(taosArrayPush(schemas, &pSW));
  pSW = NULL;

  taosMemoryFreeClear(block);

END:
  if (code != 0) {
    tqError("processBuildNew failed, code:%d", code);
  }
  tDeleteSchemaWrapper(pSW);
  blockDataFreeRes(block);
  taosMemoryFree(block);
  return code;
}
static int32_t tqProcessColData(STqReader* pReader, SSubmitTbData* pSubmitTbData, SArray* blocks, SArray* schemas) {
  int32_t code = 0;
  int32_t curRow = 0;
  int32_t lastRow = 0;

  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
  char*           assigned = taosMemoryCalloc(1, pSchemaWrapper->nCols);
  TQ_NULL_GO_TO_END(assigned);

  SArray*   pCols = pSubmitTbData->aCol;
  SColData* pCol = taosArrayGet(pCols, 0);
  TQ_NULL_GO_TO_END(pCol);
  int32_t numOfRows = pCol->nVal;
  int32_t numOfCols = taosArrayGetSize(pCols);
  tqTrace("vgId:%d, tqProcessColData start, col num: %d, rows:%d", pReader->pWalReader->pWal->cfg.vgId, numOfCols,
          numOfRows);
  for (int32_t i = 0; i < numOfRows; i++) {
    bool buildNew = false;

    for (int32_t j = 0; j < pSchemaWrapper->nCols; j++) {
      int32_t k = 0;
      for (; k < numOfCols; k++) {
        pCol = taosArrayGet(pCols, k);
        TQ_NULL_GO_TO_END(pCol);
        if (pSchemaWrapper->pSchema[j].colId == pCol->cid) {
          SColVal colVal = {0};
          TQ_ERR_GO_TO_END(tColDataGetValue(pCol, i, &colVal));
          PROCESS_VAL
          tqTrace("assign[%d] = %d, nCols:%d", j, assigned[j], numOfCols);
          break;
        }
      }
      if (k >= numOfCols) {
        // this column is not in the current row, so we set it to NULL
        assigned[j] = 0;
        buildNew = true;
      }
    }

    if (buildNew) {
      TQ_ERR_GO_TO_END(processBuildNew(pReader, pSubmitTbData, blocks, schemas, assigned, numOfRows, curRow, &lastRow));
    }

    SSDataBlock* pBlock = taosArrayGetLast(blocks);
    TQ_NULL_GO_TO_END(pBlock);

    tqTrace("vgId:%d, taosx scan, block num: %d", pReader->pWalReader->pWal->cfg.vgId,
            (int32_t)taosArrayGetSize(blocks));

    int32_t targetIdx = 0;
    int32_t sourceIdx = 0;
    int32_t colActual = blockDataGetNumOfCols(pBlock);
    while (targetIdx < colActual && sourceIdx < numOfCols) {
      pCol = taosArrayGet(pCols, sourceIdx);
      TQ_NULL_GO_TO_END(pCol);
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
      TQ_NULL_GO_TO_END(pColData);
      SColVal colVal = {0};
      TQ_ERR_GO_TO_END(tColDataGetValue(pCol, i, &colVal));
      SET_DATA
      tqTrace("targetIdx:%d sourceIdx:%d colActual:%d", targetIdx, sourceIdx, colActual);
    }

    curRow++;
  }
  SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
  pLastBlock->info.rows = curRow - lastRow;
  tqTrace("vgId:%d, tqProcessColData end, col num: %d, rows:%d, block num:%d", pReader->pWalReader->pWal->cfg.vgId,
          numOfCols, numOfRows, (int)taosArrayGetSize(blocks));
END:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("vgId:%d, process col data failed, code:%d", pReader->pWalReader->pWal->cfg.vgId, code);
  }
  taosMemoryFree(assigned);
  return code;
}

int32_t tqProcessRowData(STqReader* pReader, SSubmitTbData* pSubmitTbData, SArray* blocks, SArray* schemas) {
  int32_t   code = 0;
  STSchema* pTSchema = NULL;

  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
  char*           assigned = taosMemoryCalloc(1, pSchemaWrapper->nCols);
  TQ_NULL_GO_TO_END(assigned);

  int32_t curRow = 0;
  int32_t lastRow = 0;
  SArray* pRows = pSubmitTbData->aRowP;
  int32_t numOfRows = taosArrayGetSize(pRows);
  pTSchema = tBuildTSchema(pSchemaWrapper->pSchema, pSchemaWrapper->nCols, pSchemaWrapper->version);
  TQ_NULL_GO_TO_END(pTSchema);
  tqTrace("vgId:%d, tqProcessRowData start, rows:%d", pReader->pWalReader->pWal->cfg.vgId, numOfRows);

  for (int32_t i = 0; i < numOfRows; i++) {
    bool  buildNew = false;
    SRow* pRow = taosArrayGetP(pRows, i);
    TQ_NULL_GO_TO_END(pRow);

    for (int32_t j = 0; j < pTSchema->numOfCols; j++) {
      SColVal colVal = {0};
      TQ_ERR_GO_TO_END(tRowGet(pRow, pTSchema, j, &colVal));
      PROCESS_VAL
      tqTrace("assign[%d] = %d, nCols:%d", j, assigned[j], pTSchema->numOfCols);
    }

    if (buildNew) {
      TQ_ERR_GO_TO_END(processBuildNew(pReader, pSubmitTbData, blocks, schemas, assigned, numOfRows, curRow, &lastRow));
    }

    SSDataBlock* pBlock = taosArrayGetLast(blocks);
    TQ_NULL_GO_TO_END(pBlock);

    tqTrace("vgId:%d, taosx scan, block num: %d", pReader->pWalReader->pWal->cfg.vgId,
            (int32_t)taosArrayGetSize(blocks));

    int32_t targetIdx = 0;
    int32_t sourceIdx = 0;
    int32_t colActual = blockDataGetNumOfCols(pBlock);
    while (targetIdx < colActual && sourceIdx < pTSchema->numOfCols) {
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
      TQ_NULL_GO_TO_END(pColData);
      SColVal          colVal = {0};
      TQ_ERR_GO_TO_END(tRowGet(pRow, pTSchema, sourceIdx, &colVal));
      SET_DATA
      tqTrace("targetIdx:%d sourceIdx:%d colActual:%d", targetIdx, sourceIdx, colActual);
    }

    curRow++;
  }
  SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
  if (pLastBlock != NULL) {
    pLastBlock->info.rows = curRow - lastRow;
  }

  tqTrace("vgId:%d, tqProcessRowData end, rows:%d, block num:%d", pReader->pWalReader->pWal->cfg.vgId, numOfRows,
          (int)taosArrayGetSize(blocks));
END:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("vgId:%d, process row data failed, code:%d", pReader->pWalReader->pWal->cfg.vgId, code);
  }
  taosMemoryFreeClear(pTSchema);
  taosMemoryFree(assigned);
  return code;
}

static int32_t buildCreateTbInfo(SMqDataRsp* pRsp, SVCreateTbReq* pCreateTbReq) {
  int32_t code = 0;
  int32_t lino = 0;
  void*   createReq = NULL;
  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pCreateTbReq, code, lino, END, TSDB_CODE_INVALID_PARA);

  if (pRsp->createTableNum == 0) {
    pRsp->createTableLen = taosArrayInit(0, sizeof(int32_t));
    TSDB_CHECK_NULL(pRsp->createTableLen, code, lino, END, terrno);
    pRsp->createTableReq = taosArrayInit(0, sizeof(void*));
    TSDB_CHECK_NULL(pRsp->createTableReq, code, lino, END, terrno);
  }

  uint32_t len = 0;
  tEncodeSize(tEncodeSVCreateTbReq, pCreateTbReq, len, code);
  TSDB_CHECK_CODE(code, lino, END);
  createReq = taosMemoryCalloc(1, len);
  TSDB_CHECK_NULL(createReq, code, lino, END, terrno);

  SEncoder encoder = {0};
  tEncoderInit(&encoder, createReq, len);
  code = tEncodeSVCreateTbReq(&encoder, pCreateTbReq);
  tEncoderClear(&encoder);
  TSDB_CHECK_CODE(code, lino, END);
  TSDB_CHECK_NULL(taosArrayPush(pRsp->createTableLen, &len), code, lino, END, terrno);
  TSDB_CHECK_NULL(taosArrayPush(pRsp->createTableReq, &createReq), code, lino, END, terrno);
  pRsp->createTableNum++;
  tqTrace("build create table info msg success");

END:
  if (code != 0) {
    tqError("%s failed at %d, failed to build create table info msg:%s", __FUNCTION__, lino, tstrerror(code));
    taosMemoryFree(createReq);
  }
  return code;
}

int32_t tqRetrieveTaosxBlock(STqReader* pReader, SMqDataRsp* pRsp, SArray* blocks, SArray* schemas,
                             SSubmitTbData** pSubmitTbDataRet, SArray* rawList, int8_t fetchMeta) {
  tqTrace("tq reader retrieve data block msg pointer:%p, index:%d", pReader->msg.msgStr, pReader->nextBlk);
  SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
  if (pSubmitTbData == NULL) {
    return terrno;
  }
  pReader->nextBlk++;

  if (pSubmitTbDataRet) {
    *pSubmitTbDataRet = pSubmitTbData;
  }

  if (fetchMeta == ONLY_META) {
    if (pSubmitTbData->pCreateTbReq != NULL) {
      if (pRsp->createTableReq == NULL) {
        pRsp->createTableReq = taosArrayInit(0, POINTER_BYTES);
        if (pRsp->createTableReq == NULL) {
          return terrno;
        }
      }
      if (taosArrayPush(pRsp->createTableReq, &pSubmitTbData->pCreateTbReq) == NULL) {
        return terrno;
      }
      pSubmitTbData->pCreateTbReq = NULL;
    }
    return 0;
  }

  int32_t sversion = pSubmitTbData->sver;
  int64_t uid = pSubmitTbData->uid;
  pReader->lastBlkUid = uid;

  tDeleteSchemaWrapper(pReader->pSchemaWrapper);
  taosMemoryFreeClear(pReader->extSchema);
  pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnode->pMeta, uid, sversion, 1, &pReader->extSchema, 0);
  if (pReader->pSchemaWrapper == NULL) {
    tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
           pReader->pWalReader->pWal->cfg.vgId, uid, pReader->cachedSchemaVer);
    pReader->cachedSchemaSuid = 0;
    return TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
  }

  if (pSubmitTbData->pCreateTbReq != NULL) {
    int32_t code = buildCreateTbInfo(pRsp, pSubmitTbData->pCreateTbReq);
    if (code != 0) {
      return code;
    }
  } else if (rawList != NULL) {
    if (taosArrayPush(schemas, &pReader->pSchemaWrapper) == NULL) {
      return terrno;
    }
    pReader->pSchemaWrapper = NULL;
    return 0;
  }

  if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
    return tqProcessColData(pReader, pSubmitTbData, blocks, schemas);
  } else {
    return tqProcessRowData(pReader, pSubmitTbData, blocks, schemas);
  }
}

int32_t tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList, const char* id) {
  if (pReader == NULL || tbUidList == NULL) {
    return TSDB_CODE_SUCCESS;
  }
  if (pReader->tbIdHash) {
    taosHashClear(pReader->tbIdHash);
  } else {
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
    if (pReader->tbIdHash == NULL) {
      tqError("s-task:%s failed to init hash table", id);
      return terrno;
    }
  }

  for (int i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
    if (pKey && taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0) != 0) {
      tqError("s-task:%s failed to add table uid:%" PRId64 " to hash", id, *pKey);
      continue;
    }
  }

  tqDebug("s-task:%s %d tables are set to be queried target table", id, (int32_t)taosArrayGetSize(tbUidList));
  return TSDB_CODE_SUCCESS;
}

void tqReaderAddTbUidList(STqReader* pReader, const SArray* pTableUidList) {
  if (pReader == NULL || pTableUidList == NULL) {
    return;
  }
  if (pReader->tbIdHash == NULL) {
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
    if (pReader->tbIdHash == NULL) {
      tqError("failed to init hash table");
      return;
    }
  }

  int32_t numOfTables = taosArrayGetSize(pTableUidList);
  for (int i = 0; i < numOfTables; i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(pTableUidList, i);
    if (taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0) != 0) {
      tqError("failed to add table uid:%" PRId64 " to hash", *pKey);
      continue;
    }
    tqDebug("%s add table uid:%" PRId64 " to hash", __func__, *pKey);
  }
}

bool tqReaderIsQueriedTable(STqReader* pReader, uint64_t uid) {
  if (pReader == NULL) {
    return false;
  }
  return taosHashGet(pReader->tbIdHash, &uid, sizeof(uint64_t)) != NULL;
}

bool tqCurrentBlockConsumed(const STqReader* pReader) {
  if (pReader == NULL) {
    return false;
  }
  return pReader->msg.msgStr == NULL;
}

void tqReaderRemoveTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader == NULL || tbUidList == NULL) {
    return;
  }
  for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
    int32_t code = taosHashRemove(pReader->tbIdHash, pKey, sizeof(int64_t));
    if (code != 0) {
      tqWarn("%s failed to remove table uid:%" PRId64 " from hash, msg:%s", __func__, pKey != NULL ? *pKey : 0, tstrerror(code));
    }
  }
}

int32_t tqDeleteTbUidList(STQ* pTq, SArray* tbUidList) {
  if (pTq == NULL) {
    return 0;  // mounted vnode may have no tq
  }
  if (tbUidList == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  void*   pIter = NULL;
  int32_t vgId = TD_VID(pTq->pVnode);

  // update the table list for each consumer handle
  taosWLockLatch(&pTq->lock);
  while (1) {
    pIter = taosHashIterate(pTq->pHandle, pIter);
    if (pIter == NULL) {
      break;
    }

    STqHandle* pTqHandle = (STqHandle*)pIter;
    tqDebug("%s subKey:%s, consumer:0x%" PRIx64 " delete table list", __func__, pTqHandle->subKey, pTqHandle->consumerId);
    if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
      int32_t code = qDeleteTableListForTmqScanner(pTqHandle->execHandle.task, tbUidList);
      if (code != 0) {
        tqError("update qualified table error for %s", pTqHandle->subKey);
        continue;
      }
    } else if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
      int32_t sz = taosArrayGetSize(tbUidList);
      for (int32_t i = 0; i < sz; i++) {
        int64_t* tbUid = (int64_t*)taosArrayGet(tbUidList, i);
        if (tbUid &&
            taosHashPut(pTqHandle->execHandle.execDb.pFilterOutTbUid, tbUid, sizeof(int64_t), NULL, 0) != 0) {
          tqError("failed to add table uid:%" PRId64 " to hash", *tbUid);
          continue;
        }
      }
    } else if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
      tqReaderRemoveTbUidList(pTqHandle->execHandle.pTqReader, tbUidList);
    }
  }
  taosWUnLockLatch(&pTq->lock);
  return 0;
}

static SArray* copyUidList(const SArray* tbUidList) {
  SArray* tbUidListCopy = taosArrayInit(4, sizeof(int64_t));
  if (tbUidListCopy == NULL) {
    return NULL;
  }

  if (taosArrayAddAll(tbUidListCopy, tbUidList) == NULL) {
    taosArrayDestroy(tbUidListCopy);
    tqError("copy table uid list failed");
    return NULL;
  }
  return tbUidListCopy;
}

static int32_t addTableListForStableTmq(STqHandle* pTqHandle, STQ* pTq, const SArray* tbUidList) {
  int32_t code = 0;
  SArray* tbUidListCopy = copyUidList(tbUidList);
  if (tbUidListCopy == NULL) {
    code = terrno;
    goto END;
  }
  code = qFilterTableList(pTq->pVnode, tbUidListCopy, pTqHandle->execHandle.execTb.node,
                      pTqHandle->execHandle.task, pTqHandle->execHandle.execTb.suid);
  if (code != TDB_CODE_SUCCESS) {
    tqError("tqAddTbUidList error:%d handle %s consumer:0x%" PRIx64, code, pTqHandle->subKey,
            pTqHandle->consumerId);
    goto END;
  }
  tqDebug("%s handle %s consumer:0x%" PRIx64 " add %d tables to tqReader", __func__, pTqHandle->subKey,
          pTqHandle->consumerId, (int32_t)taosArrayGetSize(tbUidListCopy));
  tqReaderAddTbUidList(pTqHandle->execHandle.pTqReader, tbUidListCopy);

END:
  taosArrayDestroy(tbUidListCopy);
  return code;
}

int32_t tqAddTbUidList(STQ* pTq, const SArray* tbUidList) {
  if (pTq == NULL) {
    return 0;  // mounted vnode may have no tq
  }
  if (tbUidList == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  void*   pIter = NULL;
  int32_t vgId = TD_VID(pTq->pVnode);
  int32_t code = 0;

  // update the table list for each consumer handle
  taosWLockLatch(&pTq->lock);
  while (1) {
    pIter = taosHashIterate(pTq->pHandle, pIter);
    if (pIter == NULL) {
      break;
    }

    STqHandle* pTqHandle = (STqHandle*)pIter;
    tqDebug("%s subKey:%s, consumer:0x%" PRIx64 " add table list", __func__, pTqHandle->subKey, pTqHandle->consumerId);
    if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
      code = qAddTableListForTmqScanner(pTqHandle->execHandle.task, tbUidList);
      if (code != 0) {
        tqError("add table list for query tmq error for %s, msg:%s", pTqHandle->subKey, tstrerror(code));
        break;
      }
    } else if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
      code = addTableListForStableTmq(pTqHandle, pTq, tbUidList);
      if (code != 0) {
        tqError("add table list for stable tmq error for %s, msg:%s", pTqHandle->subKey, tstrerror(code));
        break;
      }
    }
  }
  taosHashCancelIterate(pTq->pHandle, pIter);
  taosWUnLockLatch(&pTq->lock);

  return code;
}

int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, SArray* cidList, SArray* cidListArray) {
  if (pTq == NULL) {
    return 0;  // mounted vnode may have no tq
  }
  if (tbUidList == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  void*   pIter = NULL;
  int32_t vgId = TD_VID(pTq->pVnode);
  int32_t code = 0;
  // update the table list for each consumer handle
  taosWLockLatch(&pTq->lock);
  while (1) {
    pIter = taosHashIterate(pTq->pHandle, pIter);
    if (pIter == NULL) {
      break;
    }

    STqHandle* pTqHandle = (STqHandle*)pIter;
    tqDebug("%s subKey:%s, consumer:0x%" PRIx64 " update table list", __func__, pTqHandle->subKey, pTqHandle->consumerId);
    if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
      SNode* pTagCond = getTagCondNodeForQueryTmq(pTqHandle->execHandle.task);
      bool ret = checkCidInTagCondition(pTagCond, cidList);
      if (ret){
        code = qUpdateTableListForTmqScanner(pTqHandle->execHandle.task, tbUidList);
        if (code != 0) {
          tqError("update table list for query tmq error for %s, msg:%s", pTqHandle->subKey, tstrerror(code));
          break;
        }
      }
      qUpdateTableTagCacheForTmq(pTqHandle->execHandle.task, tbUidList, cidList, cidListArray);
    } else if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
      SNode* pTagCond = getTagCondNodeForStableTmq(pTqHandle->execHandle.execTb.node);
      bool ret = checkCidInTagCondition(pTagCond, cidList);
      if (ret){
        tqReaderRemoveTbUidList(pTqHandle->execHandle.pTqReader, tbUidList);
        code = addTableListForStableTmq(pTqHandle, pTq, tbUidList);
        if (code != 0) {
          tqError("update table list for stable tmq error for %s, msg:%s", pTqHandle->subKey, tstrerror(code));
          break;
        }
      }
    }
  }

  taosHashCancelIterate(pTq->pHandle, pIter);
  taosWUnLockLatch(&pTq->lock);

  return code;
}

static void destroySourceScanTables(void* ptr) {
  SArray** pTables = ptr;
  if (pTables && *pTables) {
    taosArrayDestroy(*pTables);
    *pTables = NULL;
  }
}

static int32_t compareSVTColInfo(const void* p1, const void* p2) {
  SVTColInfo* pCol1 = (SVTColInfo*)p1;
  SVTColInfo* pCol2 = (SVTColInfo*)p2;
  if (pCol1->vColId == pCol2->vColId) {
    return 0;
  } else if (pCol1->vColId < pCol2->vColId) {
    return -1;
  } else {
    return 1;
  }
}

static void freeTableSchemaCache(const void* key, size_t keyLen, void* value, void* ud) {
  if (value) {
    SSchemaWrapper* pSchemaWrapper = value;
    tDeleteSchemaWrapper(pSchemaWrapper);
  }
}

taosdata / TDengine / #5011

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous