#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

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

48.47

/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 "tmsg.h"
#include "tq.h"

static int32_t tqCollectPhysicalTables(STqReader* pReader, const char* idstr);

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

  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) {
    SVCreateTbBatchReq req = {0};
    if (tDecodeSVCreateTbBatchReq(&dcoder, &req) < 0) {
      goto end;
    }

    int32_t        needRebuild = 0;
    SVCreateTbReq* pCreateReq = NULL;
    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
      pCreateReq = req.pReqs + iReq;
      if (pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid) {
        needRebuild++;
      }
    }
    if (needRebuild == 0) {
      // do nothing
    } else if (needRebuild == req.nReqs) {
      realTbSuid = tbSuid;
    } else {
      realTbSuid = tbSuid;
      SVCreateTbBatchReq reqNew = {0};
      reqNew.pArray = taosArrayInit(req.nReqs, sizeof(struct SVCreateTbReq));
      if (reqNew.pArray == NULL) {
        tDeleteSVCreateTbBatchReq(&req);
        goto end;
      }
      for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
        pCreateReq = req.pReqs + iReq;
        if (pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid) {
          reqNew.nReqs++;
          if (taosArrayPush(reqNew.pArray, pCreateReq) == NULL) {
            taosArrayDestroy(reqNew.pArray);
            tDeleteSVCreateTbBatchReq(&req);
            goto end;
          }
        }
      }

      int     tlen = 0;
      int32_t ret = 0;
      tEncodeSize(tEncodeSVCreateTbBatchReq, &reqNew, tlen, ret);
      void* buf = taosMemoryMalloc(tlen);
      if (NULL == buf) {
        taosArrayDestroy(reqNew.pArray);
        tDeleteSVCreateTbBatchReq(&req);
        goto end;
      }
      SEncoder coderNew = {0};
      tEncoderInit(&coderNew, buf, tlen - sizeof(SMsgHead));
      ret = tEncodeSVCreateTbBatchReq(&coderNew, &reqNew);
      tEncoderClear(&coderNew);
      if (ret < 0) {
        taosMemoryFree(buf);
        taosArrayDestroy(reqNew.pArray);
        tDeleteSVCreateTbBatchReq(&req);
        goto end;
      }
      (void)memcpy(pHead->body + sizeof(SMsgHead), buf, tlen);
      pHead->bodyLen = tlen + sizeof(SMsgHead);
      taosMemoryFree(buf);
      taosArrayDestroy(reqNew.pArray);
    }

    tDeleteSVCreateTbBatchReq(&req);
  } else if (msgType == TDMT_VND_ALTER_TABLE) {
    SVAlterTbReq req = {0};

    if (tDecodeSVAlterTbReq(&dcoder, &req) < 0) {
      goto end;
    }

    SMetaReader mr = {0};
    metaReaderDoInit(&mr, pHandle->execHandle.pTqReader->pVnodeMeta, META_READER_LOCK);

    if (metaGetTableEntryByName(&mr, req.tbName) < 0) {
      metaReaderClear(&mr);
      goto end;
    }
    realTbSuid = mr.me.ctbEntry.suid;
    metaReaderClear(&mr);
  } else if (msgType == TDMT_VND_DROP_TABLE) {
    SVDropTbBatchReq req = {0};

    if (tDecodeSVDropTbBatchReq(&dcoder, &req) < 0) {
      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) {
        needRebuild++;
      }
    }
    if (needRebuild == 0) {
      // do nothing
    } else if (needRebuild == req.nReqs) {
      realTbSuid = tbSuid;
    } else {
      realTbSuid = tbSuid;
      SVDropTbBatchReq reqNew = {0};
      reqNew.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbReq));
      if (reqNew.pArray == NULL) {
        goto end;
      }
      for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
        pDropReq = req.pReqs + iReq;
        if (pDropReq->suid == tbSuid) {
          reqNew.nReqs++;
          if (taosArrayPush(reqNew.pArray, pDropReq) == NULL) {
            taosArrayDestroy(reqNew.pArray);
            goto end;
          }
        }
      }

      int     tlen = 0;
      int32_t ret = 0;
      tEncodeSize(tEncodeSVDropTbBatchReq, &reqNew, tlen, ret);
      void* buf = taosMemoryMalloc(tlen);
      if (NULL == buf) {
        taosArrayDestroy(reqNew.pArray);
        goto end;
      }
      SEncoder coderNew = {0};
      tEncoderInit(&coderNew, buf, tlen - sizeof(SMsgHead));
      ret = tEncodeSVDropTbBatchReq(&coderNew, &reqNew);
      tEncoderClear(&coderNew);
      if (ret != 0) {
        taosMemoryFree(buf);
        taosArrayDestroy(reqNew.pArray);
        goto end;
      }
      (void)memcpy(pHead->body + sizeof(SMsgHead), buf, tlen);
      pHead->bodyLen = tlen + sizeof(SMsgHead);
      taosMemoryFree(buf);
      taosArrayDestroy(reqNew.pArray);
    }
  } 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->pVnodeMeta, uid, -1, 1, NULL);
  if (schema && schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY) {
    ret = true;
  }
  tDeleteSchemaWrapper(schema);
  pReader->hasPrimaryKey = ret;
}

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, NULL, 0);
  if (pReader->pWalReader == NULL) {
    taosMemoryFree(pReader);
    return NULL;
  }

  pReader->pVnodeMeta = pVnode->pMeta;
  pReader->pColIdList = NULL;
  pReader->cachedSchemaVer = 0;
  pReader->cachedSchemaSuid = 0;
  pReader->pSchemaWrapper = NULL;
  pReader->tbIdHash = NULL;
  pReader->pResBlock = NULL;

  int32_t code = createDataBlock(&pReader->pResBlock);
  if (code) {
    terrno = code;
  }

  return pReader;
}

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

  // close wal reader
  if (pReader->pWalReader) {
    walCloseReader(pReader->pWalReader);
  }

  if (pReader->pSchemaWrapper) {
    tDeleteSchemaWrapper(pReader->pSchemaWrapper);
  }

  taosMemoryFree(pReader->extSchema);
  if (pReader->pColIdList) {
    taosArrayDestroy(pReader->pColIdList);
  }

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

  taosHashCleanup(pReader->vtSourceScanInfo.pVirtualTables);
  taosHashCleanup(pReader->vtSourceScanInfo.pPhysicalTables);
  taosLRUCacheCleanup(pReader->vtSourceScanInfo.pPhyTblSchemaCache);
  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;
}

int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, int64_t maxVer, const char* id) {
  int32_t code = 0;

  while (1) {
    TAOS_CHECK_RETURN(walNextValidMsg(pReader));

    SWalCont* pCont = &pReader->pHead->head;
    int64_t   ver = pCont->version;
    if (ver > maxVer) {
      tqDebug("maxVer in WAL:%" PRId64 " reached, current:%" PRId64 ", do not scan wal anymore, %s", maxVer, ver, id);
      return TSDB_CODE_SUCCESS;
    }

    if (pCont->msgType == TDMT_VND_SUBMIT) {
      void*   pBody = POINTER_SHIFT(pCont->body, sizeof(SSubmitReq2Msg));
      int32_t len = pCont->bodyLen - sizeof(SSubmitReq2Msg);

      void* data = taosMemoryMalloc(len);
      if (data == NULL) {
        // todo: for all stream in this vnode, keep this offset in the offset files, and wait for a moment, and then
        // retry
        tqError("vgId:%d, failed to copy submit data for stream processing, since out of memory", 0);
        return terrno;
      }

      (void)memcpy(data, pBody, len);
      SPackedData data1 = (SPackedData){.ver = ver, .msgLen = len, .msgStr = data};

      code = streamDataSubmitNew(&data1, STREAM_INPUT__DATA_SUBMIT, (SStreamDataSubmit**)pItem);
      if (code != 0) {
        tqError("%s failed to create data submit for stream since out of memory", id);
        return code;
      }
    } else if (pCont->msgType == TDMT_VND_DELETE) {
      void*       pBody = POINTER_SHIFT(pCont->body, sizeof(SMsgHead));
      int32_t     len = pCont->bodyLen - sizeof(SMsgHead);
      EStreamType blockType = STREAM_DELETE_DATA;
      code = tqExtractDelDataBlock(pBody, len, ver, (void**)pItem, 0, blockType);
      if (code == TSDB_CODE_SUCCESS) {
        if (*pItem == NULL) {
          tqDebug("s-task:%s empty delete msg, discard it, len:%d, ver:%" PRId64, id, len, ver);
          // we need to continue check next data in the wal files.
          continue;
        } else {
          tqDebug("s-task:%s delete msg extract from WAL, len:%d, ver:%" PRId64, id, len, ver);
        }
      } else {
        terrno = code;
        tqError("s-task:%s extract delete msg from WAL failed, code:%s", id, tstrerror(code));
        return code;
      }

    } else if (pCont->msgType == TDMT_VND_DROP_TABLE && pReader->cond.scanDropCtb) {
      void*   pBody = POINTER_SHIFT(pCont->body, sizeof(SMsgHead));
      int32_t len = pCont->bodyLen - sizeof(SMsgHead);
      code = tqExtractDropCtbDataBlock(pBody, len, ver, (void**)pItem, 0);
      if (TSDB_CODE_SUCCESS == code) {
        if (!*pItem) {
          continue;
        } else {
          tqDebug("s-task:%s drop ctb msg extract from WAL, len:%d, ver:%" PRId64, id, len, ver);
        }
      } else {
        terrno = code;
        return code;
      }
    } else {
      tqError("s-task:%s invalid msg type:%d, ver:%" PRId64, id, pCont->msgType, ver);
      return TSDB_CODE_STREAM_INTERNAL_ERROR;
    }

    return code;
  }
}

bool tqNextBlockInWal(STqReader* pReader, const char* id, int sourceExcluded) {
  if (pReader == NULL) {
    return false;
  }
  SWalReader* pWalReader = pReader->pWalReader;

  int64_t st = taosGetTimestampMs();
  while (1) {
    int32_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData);
    while (pReader->nextBlk < numOfBlocks) {
      tqTrace("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 false;
      }
      if ((pSubmitTbData->flags & sourceExcluded) != 0) {
        pReader->nextBlk += 1;
        continue;
      }
      if (pReader->tbIdHash == NULL || taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t)) != NULL) {
        tqTrace("tq reader return submit block, uid:%" PRId64, pSubmitTbData->uid);
        SSDataBlock* pRes = NULL;
        int32_t      code = tqRetrieveDataBlock(pReader, &pRes, NULL);
        if (code == TSDB_CODE_SUCCESS) {
          return true;
        }
      } else {
        pReader->nextBlk += 1;
        tqTrace("tq reader discard submit block, uid:%" PRId64 ", continue", pSubmitTbData->uid);
      }
    }

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

    int64_t elapsed = taosGetTimestampMs() - st;
    if (elapsed > 1000 || elapsed < 0) {
      return false;
    }

    // try next message in wal file
    if (walNextValidMsg(pWalReader) < 0) {
      return false;
    }

    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;
    if (tqReaderSetSubmitMsg(pReader, pBody, bodyLen, ver, NULL) != 0) {
      return false;
    }
    pReader->nextBlk = 0;
  }
}

int32_t tqReaderSetSubmitMsg(STqReader* pReader, void* msgStr, int32_t msgLen, int64_t ver, SArray* rawList) {
  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);
  SDecoder decoder = {0};

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

  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;
}

SSDataBlock* tqGetResultBlock(STqReader* pReader) {
  if (pReader == NULL) {
    return NULL;
  }
  return pReader->pResBlock;
}

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 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 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 buildResSDataBlock(STqReader* pReader, SSchemaWrapper* pSchema, const SArray* pColIdList) {
  if (pReader == NULL || pSchema == NULL || pColIdList == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  SSDataBlock* pBlock = pReader->pResBlock;
  if (blockDataGetNumOfCols(pBlock) > 0) {
    blockDataDestroy(pBlock);
    int32_t code = createDataBlock(&pReader->pResBlock);
    if (code) {
      return code;
    }
    pBlock = pReader->pResBlock;

    pBlock->info.id.uid = pReader->cachedSchemaUid;
    pBlock->info.version = pReader->msg.ver;
  }

  int32_t numOfCols = taosArrayGetSize(pColIdList);

  if (numOfCols == 0) {  // all columns are required
    for (int32_t i = 0; i < pSchema->nCols; ++i) {
      SSchema*        pColSchema = &pSchema->pSchema[i];
      SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);

      if (IS_DECIMAL_TYPE(pColSchema->type) && pReader->extSchema != NULL) {
        decimalFromTypeMod(pReader->extSchema[i].typeMod, &colInfo.info.precision, &colInfo.info.scale);
      }
      int32_t code = blockDataAppendColInfo(pBlock, &colInfo);
      if (code != TSDB_CODE_SUCCESS) {
        blockDataFreeRes(pBlock);
        return terrno;
      }
    }
  } else {
    if (numOfCols > pSchema->nCols) {
      numOfCols = pSchema->nCols;
    }

    int32_t i = 0;
    int32_t j = 0;
    while (i < pSchema->nCols && j < numOfCols) {
      SSchema* pColSchema = &pSchema->pSchema[i];
      col_id_t colIdSchema = pColSchema->colId;

      col_id_t* pColIdNeed = (col_id_t*)taosArrayGet(pColIdList, j);
      if (pColIdNeed == NULL) {
        break;
      }
      if (colIdSchema < *pColIdNeed) {
        i++;
      } else if (colIdSchema > *pColIdNeed) {
        j++;
      } else {
        SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
        if (IS_DECIMAL_TYPE(pColSchema->type) && pReader->extSchema != NULL) {
          decimalFromTypeMod(pReader->extSchema[i].typeMod, &colInfo.info.precision, &colInfo.info.scale);
        }
        int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
        if (code != TSDB_CODE_SUCCESS) {
          return -1;
        }
        i++;
        j++;
      }
    }
  }

  return TSDB_CODE_SUCCESS;
}

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)) {
    char val[65535 + 2] = {0};
    if (COL_VAL_IS_VALUE(pColVal)) {
      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;
}

int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* id) {
  if (pReader == NULL || pRes == 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;
  STSchema*      pTSchema = NULL;
  SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk++);
  TSDB_CHECK_NULL(pSubmitTbData, code, line, END, terrno);
  SSDataBlock* pBlock = pReader->pResBlock;
  *pRes = pBlock;

  blockDataCleanup(pBlock);

  int32_t vgId = pReader->pWalReader->pWal->cfg.vgId;
  int32_t sversion = pSubmitTbData->sver;
  int64_t suid = pSubmitTbData->suid;
  int64_t uid = pSubmitTbData->uid;
  pReader->lastTs = pSubmitTbData->ctimeMs;

  pBlock->info.id.uid = uid;
  pBlock->info.version = pReader->msg.ver;

  if ((suid != 0 && pReader->cachedSchemaSuid != suid) || (suid == 0 && pReader->cachedSchemaUid != uid) ||
      (pReader->cachedSchemaVer != sversion)) {
    tDeleteSchemaWrapper(pReader->pSchemaWrapper);
    taosMemoryFree(pReader->extSchema);
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1, &pReader->extSchema);
    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);
      pReader->cachedSchemaSuid = 0;
      return TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
    }

    pReader->cachedSchemaUid = uid;
    pReader->cachedSchemaSuid = suid;
    pReader->cachedSchemaVer = sversion;

    if (pReader->cachedSchemaVer != pReader->pSchemaWrapper->version) {
      tqError("vgId:%d, schema version mismatch, suid:%" PRId64 ", uid:%" PRId64 ", version:%d, cached version:%d",
              vgId, suid, uid, sversion, pReader->pSchemaWrapper->version);
      return TSDB_CODE_TQ_INTERNAL_ERROR;
    }
    code = buildResSDataBlock(pReader, pReader->pSchemaWrapper, pReader->pColIdList);
    TSDB_CHECK_CODE(code, line, END);
    pBlock = pReader->pResBlock;
    *pRes = pBlock;
  }

  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, numOfRows);
  TSDB_CHECK_CODE(code, line, END);
  pBlock->info.rows = numOfRows;
  int32_t colActual = blockDataGetNumOfCols(pBlock);

  // convert and scan one block
  if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
    SArray* pCols = pSubmitTbData->aCol;
    int32_t numOfCols = taosArrayGetSize(pCols);
    int32_t targetIdx = 0;
    int32_t sourceIdx = 0;
    while (targetIdx < colActual) {
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
      TSDB_CHECK_NULL(pColData, code, line, END, terrno);
      if (sourceIdx >= numOfCols) {
        tqError("lostdata tqRetrieveDataBlock sourceIdx:%d >= numOfCols:%d", sourceIdx, numOfCols);
        colDataSetNNULL(pColData, 0, numOfRows);
        targetIdx++;
        continue;
      }

      SColData* pCol = taosArrayGet(pCols, sourceIdx);
      TSDB_CHECK_NULL(pCol, code, line, END, terrno);
      SColVal colVal = {0};
      tqTrace("lostdata colActual:%d, sourceIdx:%d, targetIdx:%d, numOfCols:%d, source cid:%d, dst cid:%d", colActual,
              sourceIdx, targetIdx, numOfCols, pCol->cid, pColData->info.colId);
      if (pCol->cid < pColData->info.colId) {
        sourceIdx++;
      } else if (pCol->cid == pColData->info.colId) {
        for (int32_t i = 0; i < pCol->nVal; i++) {
          code = tColDataGetValue(pCol, i, &colVal);
          TSDB_CHECK_CODE(code, line, END);
          code = doSetVal(pColData, i, &colVal);
          TSDB_CHECK_CODE(code, line, END);
        }
        sourceIdx++;
        targetIdx++;
      } else {
        colDataSetNNULL(pColData, 0, numOfRows);
        targetIdx++;
      }
    }
  } else {
    SArray*         pRows = pSubmitTbData->aRowP;
    SSchemaWrapper* pWrapper = pReader->pSchemaWrapper;
    pTSchema = tBuildTSchema(pWrapper->pSchema, pWrapper->nCols, pWrapper->version);
    TSDB_CHECK_NULL(pTSchema, code, line, END, terrno);

    for (int32_t i = 0; i < numOfRows; i++) {
      SRow* pRow = taosArrayGetP(pRows, i);
      TSDB_CHECK_NULL(pRow, code, line, END, terrno);
      int32_t sourceIdx = 0;
      for (int32_t j = 0; j < colActual; j++) {
        SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, j);
        TSDB_CHECK_NULL(pColData, code, line, END, terrno);

        while (1) {
          SColVal colVal = {0};
          code = tRowGet(pRow, pTSchema, sourceIdx, &colVal);
          TSDB_CHECK_CODE(code, line, END);

          if (colVal.cid < pColData->info.colId) {
            sourceIdx++;
            continue;
          } else if (colVal.cid == pColData->info.colId) {
            code = doSetVal(pColData, i, &colVal);
            TSDB_CHECK_CODE(code, line, END);
            sourceIdx++;
            break;
          } else {
            colDataSetNULL(pColData, i);
            break;
          }
        }
      }
    }
  }

END:
  if (code != 0) {
    tqError("tqRetrieveDataBlock failed, line:%d, msg:%s", line, tstrerror(code));
  }
  taosMemoryFreeClear(pTSchema);
  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) {                   \
    TQ_ERR_GO_TO_END(doSetVal(pColData, curRow - lastRow, &colVal)); \
    sourceIdx++;                                                     \
    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 < numOfCols; j++) {
      pCol = taosArrayGet(pCols, j);
      TQ_NULL_GO_TO_END(pCol);
      SColVal colVal = {0};
      TQ_ERR_GO_TO_END(tColDataGetValue(pCol, i, &colVal));
      PROCESS_VAL
    }

    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) {
      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
    }

    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
    }

    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) {
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
      SColVal          colVal = {0};
      TQ_ERR_GO_TO_END(tRowGet(pRow, pTSchema, sourceIdx, &colVal));
      SET_DATA
    }

    curRow++;
  }
  SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
  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);
  taosMemoryFree(pReader->extSchema);
  pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1, &pReader->extSchema);
  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 tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList, const char* id) {
  if (pReader == NULL) {
    return TSDB_CODE_SUCCESS;
  }
  pReader->pColIdList = pColIdList;
  return tqCollectPhysicalTables(pReader, id);
}

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 tqCollectPhysicalTables(pReader, id);
}

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;
    }
  }
}

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);
    if (pKey && taosHashRemove(pReader->tbIdHash, pKey, sizeof(int64_t)) != 0) {
      tqError("failed to remove table uid:%" PRId64 " from hash", *pKey);
    }
  }
}

int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
  if (pTq == NULL || 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;
    if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
      int32_t code = qUpdateTableListForStreamScanner(pTqHandle->execHandle.task, tbUidList, isAdd);
      if (code != 0) {
        tqError("update qualified table error for %s", pTqHandle->subKey);
        continue;
      }
    } else if (pTqHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
      if (!isAdd) {
        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) {
      if (isAdd) {
        SArray* list = NULL;
        int     ret = qGetTableList(pTqHandle->execHandle.execTb.suid, pTq->pVnode, pTqHandle->execHandle.execTb.node,
                                    &list, pTqHandle->execHandle.task);
        if (ret != TDB_CODE_SUCCESS) {
          tqError("qGetTableList in tqUpdateTbUidList error:%d handle %s consumer:0x%" PRIx64, ret, pTqHandle->subKey,
                  pTqHandle->consumerId);
          taosArrayDestroy(list);
          taosHashCancelIterate(pTq->pHandle, pIter);
          taosWUnLockLatch(&pTq->lock);

          return ret;
        }
        tqReaderSetTbUidList(pTqHandle->execHandle.pTqReader, list, NULL);
        taosArrayDestroy(list);
      } else {
        tqReaderRemoveTbUidList(pTqHandle->execHandle.pTqReader, tbUidList);
      }
    }
  }
  taosWUnLockLatch(&pTq->lock);

  // update the table list handle for each stream scanner/wal reader
  streamMetaWLock(pTq->pStreamMeta);
  while (1) {
    pIter = taosHashIterate(pTq->pStreamMeta->pTasksMap, pIter);
    if (pIter == NULL) {
      break;
    }

    int64_t      refId = *(int64_t*)pIter;
    SStreamTask* pTask = taosAcquireRef(streamTaskRefPool, refId);
    if (pTask != NULL) {
      int32_t taskId = pTask->id.taskId;

      if ((pTask->info.taskLevel == TASK_LEVEL__SOURCE) && (pTask->exec.pExecutor != NULL)) {
        int32_t code = qUpdateTableListForStreamScanner(pTask->exec.pExecutor, tbUidList, isAdd);
        if (code != 0) {
          tqError("vgId:%d, s-task:0x%x update qualified table error for stream task", vgId, taskId);
        }
      }
      int32_t ret = taosReleaseRef(streamTaskRefPool, refId);
      if (ret) {
        tqError("vgId:%d release task refId failed, refId:%" PRId64, vgId, refId);
      }
    }
  }

  streamMetaWUnLock(pTq->pStreamMeta);
  return 0;
}

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;
  }
}

int32_t tqReaderSetVtableInfo(STqReader* pReader, void* vnode, void* ptr, SSHashObj* pVtableInfos,
                              SSDataBlock** ppResBlock, const char* idstr) {
  int32_t            code = TSDB_CODE_SUCCESS;
  int32_t            lino = 0;
  SStorageAPI*       pAPI = ptr;
  SVTSourceScanInfo* pScanInfo = NULL;
  SHashObj*          pVirtualTables = NULL;
  SMetaReader        metaReader = {0};
  SVTColInfo         colInfo = {0};
  SSchemaWrapper*    schema = NULL;

  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(vnode, code, lino, _end, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pAPI, code, lino, _end, TSDB_CODE_INVALID_PARA);

  pScanInfo = &pReader->vtSourceScanInfo;
  taosHashCleanup(pScanInfo->pVirtualTables);
  pScanInfo->pVirtualTables = NULL;

  if (tSimpleHashGetSize(pVtableInfos) == 0) {
    goto _end;
  }

  pVirtualTables = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  TSDB_CHECK_NULL(pVirtualTables, code, lino, _end, terrno);
  taosHashSetFreeFp(pVirtualTables, destroySourceScanTables);

  int32_t iter = 0;
  void*   px = tSimpleHashIterate(pVtableInfos, NULL, &iter);
  while (px != NULL) {
    int64_t vTbUid = *(int64_t*)tSimpleHashGetKey(px, NULL);
    SArray* pColInfos = taosArrayInit(8, sizeof(SVTColInfo));
    TSDB_CHECK_NULL(pColInfos, code, lino, _end, terrno);
    code = taosHashPut(pVirtualTables, &vTbUid, sizeof(int64_t), &pColInfos, POINTER_BYTES);
    TSDB_CHECK_CODE(code, lino, _end);

    SSHashObj* pPhysicalTables = *(SSHashObj**)px;
    int32_t    iterIn = 0;
    void*      pxIn = tSimpleHashIterate(pPhysicalTables, NULL, &iterIn);
    while (pxIn != NULL) {
      char* physicalTableName = tSimpleHashGetKey(pxIn, NULL);
      pAPI->metaReaderFn.clearReader(&metaReader);
      pAPI->metaReaderFn.initReader(&metaReader, vnode, META_READER_LOCK, &pAPI->metaFn);
      code = pAPI->metaReaderFn.getTableEntryByName(&metaReader, physicalTableName);
      TSDB_CHECK_CODE(code, lino, _end);
      pAPI->metaReaderFn.readerReleaseLock(&metaReader);
      colInfo.pTbUid = metaReader.me.uid;

      switch (metaReader.me.type) {
        case TSDB_CHILD_TABLE: {
          int64_t suid = metaReader.me.ctbEntry.suid;
          pAPI->metaReaderFn.clearReader(&metaReader);
          pAPI->metaReaderFn.initReader(&metaReader, vnode, META_READER_LOCK, &pAPI->metaFn);
          code = pAPI->metaReaderFn.getTableEntryByUid(&metaReader, suid);
          TSDB_CHECK_CODE(code, lino, _end);
          pAPI->metaReaderFn.readerReleaseLock(&metaReader);
          schema = &metaReader.me.stbEntry.schemaRow;
          break;
        }
        case TSDB_NORMAL_TABLE: {
          schema = &metaReader.me.ntbEntry.schemaRow;
          break;
        }
        default: {
          tqError("invalid table type: %d", metaReader.me.type);
          code = TSDB_CODE_INVALID_PARA;
          TSDB_CHECK_CODE(code, lino, _end);
        }
      }

      SArray* pCols = *(SArray**)pxIn;
      int32_t ncols = taosArrayGetSize(pCols);
      for (int32_t i = 0; i < ncols; ++i) {
        SColIdName* pCol = taosArrayGet(pCols, i);
        colInfo.vColId = pCol->colId;

        for (int32_t j = 0; j < schema->nCols; ++j) {
          if (strncmp(pCol->colName, schema->pSchema[j].name, strlen(schema->pSchema[j].name)) == 0) {
            colInfo.pColId = schema->pSchema[j].colId;
            void* px = taosArrayPush(pColInfos, &colInfo);
            TSDB_CHECK_NULL(px, code, lino, _end, terrno);
            break;
          }
        }
      }

      taosArraySort(pColInfos, compareSVTColInfo);
      pxIn = tSimpleHashIterate(pPhysicalTables, pxIn, &iterIn);
    }

    px = tSimpleHashIterate(pVtableInfos, px, &iter);
  }

  pScanInfo->pVirtualTables = pVirtualTables;
  pVirtualTables = NULL;

  // set the result data block
  if (pReader->pResBlock) {
    blockDataDestroy(pReader->pResBlock);
  }
  pReader->pResBlock = *ppResBlock;
  *ppResBlock = NULL;

  // update reader callback for vtable source scan
  pAPI->tqReaderFn.tqRetrieveBlock = tqRetrieveVTableDataBlock;
  pAPI->tqReaderFn.tqNextBlockImpl = tqNextVTableSourceBlockImpl;
  pAPI->tqReaderFn.tqReaderIsQueriedTable = tqReaderIsQueriedSourceTable;

_end:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at line %d since %s, id: %s", __func__, lino, tstrerror(code), idstr);
  }
  pAPI->metaReaderFn.clearReader(&metaReader);
  if (pVirtualTables != NULL) {
    taosHashCleanup(pVirtualTables);
  }
  return code;
}

static int32_t tqCollectPhysicalTables(STqReader* pReader, const char* idstr) {
  int32_t            code = TSDB_CODE_SUCCESS;
  int32_t            lino = 0;
  SVTSourceScanInfo* pScanInfo = NULL;
  SHashObj*          pVirtualTables = NULL;
  SHashObj*          pPhysicalTables = NULL;
  void*              pIter = NULL;
  void*              px = NULL;

  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);

  pScanInfo = &pReader->vtSourceScanInfo;
  taosHashCleanup(pScanInfo->pPhysicalTables);
  pScanInfo->pPhysicalTables = NULL;
  taosLRUCacheCleanup(pScanInfo->pPhyTblSchemaCache);
  pScanInfo->pPhyTblSchemaCache = NULL;
  pScanInfo->nextVirtualTableIdx = -1;
  pScanInfo->metaFetch = 0;
  pScanInfo->cacheHit = 0;

  pVirtualTables = pScanInfo->pVirtualTables;
  if (taosHashGetSize(pVirtualTables) == 0 || taosHashGetSize(pReader->tbIdHash) == 0 ||
      taosArrayGetSize(pReader->pColIdList) == 0) {
    goto _end;
  }

  pPhysicalTables = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
  TSDB_CHECK_NULL(pPhysicalTables, code, lino, _end, terrno);
  taosHashSetFreeFp(pPhysicalTables, destroySourceScanTables);

  pIter = taosHashIterate(pReader->tbIdHash, NULL);
  while (pIter != NULL) {
    int64_t vTbUid = *(int64_t*)taosHashGetKey(pIter, NULL);

    px = taosHashGet(pVirtualTables, &vTbUid, sizeof(int64_t));
    TSDB_CHECK_NULL(px, code, lino, _end, terrno);
    SArray* pColInfos = *(SArray**)px;
    TSDB_CHECK_NULL(pColInfos, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);

    // Traverse all required columns and collect corresponding physical tables
    int32_t nColInfos = taosArrayGetSize(pColInfos);
    int32_t nOutputCols = taosArrayGetSize(pReader->pColIdList);
    for (int32_t i = 0, j = 0; i < nColInfos && j < nOutputCols;) {
      SVTColInfo* pCol = taosArrayGet(pColInfos, i);
      col_id_t    colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, j);
      if (pCol->vColId < colIdNeed) {
        i++;
      } else if (pCol->vColId > colIdNeed) {
        j++;
      } else {
        SArray* pRelatedVTs = NULL;
        px = taosHashGet(pPhysicalTables, &pCol->pTbUid, sizeof(int64_t));
        if (px == NULL) {
          pRelatedVTs = taosArrayInit(8, sizeof(int64_t));
          TSDB_CHECK_NULL(pRelatedVTs, code, lino, _end, terrno);
          code = taosHashPut(pPhysicalTables, &pCol->pTbUid, sizeof(int64_t), &pRelatedVTs, POINTER_BYTES);
          if (code != TSDB_CODE_SUCCESS) {
            taosArrayDestroy(pRelatedVTs);
            TSDB_CHECK_CODE(code, lino, _end);
          }
        } else {
          pRelatedVTs = *(SArray**)px;
        }
        if (taosArrayGetSize(pRelatedVTs) == 0 || *(int64_t*)taosArrayGetLast(pRelatedVTs) != vTbUid) {
          px = taosArrayPush(pRelatedVTs, &vTbUid);
          TSDB_CHECK_NULL(px, code, lino, _end, terrno);
        }
        i++;
        j++;
      }
    }
    pIter = taosHashIterate(pReader->tbIdHash, pIter);
  }

  pScanInfo->pPhysicalTables = pPhysicalTables;
  pPhysicalTables = NULL;

  if (taosHashGetSize(pScanInfo->pPhysicalTables) > 0) {
    pScanInfo->pPhyTblSchemaCache = taosLRUCacheInit(1024 * 128, -1, .5);
    TSDB_CHECK_NULL(pScanInfo->pPhyTblSchemaCache, code, lino, _end, terrno);
  }

_end:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at line %d since %s, id: %s", __func__, lino, tstrerror(code), idstr);
  }
  if (pIter != NULL) {
    taosHashCancelIterate(pReader->tbIdHash, pIter);
  }
  if (pPhysicalTables != NULL) {
    taosHashCleanup(pPhysicalTables);
  }
  return code;
}

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

int32_t tqRetrieveVTableDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* idstr) {
  int32_t            code = TSDB_CODE_SUCCESS;
  int32_t            lino = 0;
  SVTSourceScanInfo* pScanInfo = NULL;
  SSubmitTbData*     pSubmitTbData = NULL;
  SSDataBlock*       pBlock = NULL;
  void*              px = NULL;
  int64_t            vTbUid = 0;
  int64_t            pTbUid = 0;
  LRUHandle*         h = NULL;
  STSchema*          pPhyTblSchema = NULL;

  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pRes, code, lino, _end, TSDB_CODE_INVALID_PARA);

  pScanInfo = &pReader->vtSourceScanInfo;
  tqDebug("tq reader retrieve vtable data block from %p, nextBlk:%d, vtbIdx:%d, id:%s", pReader->msg.msgStr,
          pReader->nextBlk, pScanInfo->nextVirtualTableIdx, idstr);

  *pRes = NULL;
  pBlock = pReader->pResBlock;
  blockDataCleanup(pBlock);

  pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
  TSDB_CHECK_NULL(pSubmitTbData, code, lino, _end, terrno);

  pReader->lastTs = pSubmitTbData->ctimeMs;

  pTbUid = pSubmitTbData->uid;
  px = taosHashGet(pScanInfo->pPhysicalTables, &pTbUid, sizeof(int64_t));
  TSDB_CHECK_NULL(px, code, lino, _end, terrno);
  SArray* pRelatedVTs = *(SArray**)px;
  vTbUid = *(int64_t*)taosArrayGet(pRelatedVTs, pScanInfo->nextVirtualTableIdx);
  px = taosHashGet(pScanInfo->pVirtualTables, &vTbUid, sizeof(int64_t));
  TSDB_CHECK_NULL(px, code, lino, _end, terrno);
  SArray* pColInfos = *(SArray**)px;
  TSDB_CHECK_NULL(pColInfos, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);

  int32_t nColInfos = taosArrayGetSize(pColInfos);
  int32_t nOutputCols = taosArrayGetSize(pBlock->pDataBlock);

  int32_t numOfRows = 0;
  int32_t nInputCols = 0;
  if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
    SColData* pCol = taosArrayGet(pSubmitTbData->aCol, 0);
    TSDB_CHECK_NULL(pCol, code, lino, _end, terrno);
    numOfRows = pCol->nVal;
    nInputCols = taosArrayGetSize(pSubmitTbData->aCol);
  } else {
    // try to get physical table schema from cache
    pScanInfo->metaFetch++;
    int64_t         cacheKey = (pSubmitTbData->suid == 0) ? pTbUid : pSubmitTbData->suid;
    SSchemaWrapper* pWrapper = NULL;
    h = taosLRUCacheLookup(pScanInfo->pPhyTblSchemaCache, &cacheKey, sizeof(int64_t));
    if (h != NULL) {
      pWrapper = taosLRUCacheValue(pScanInfo->pPhyTblSchemaCache, h);
      TSDB_CHECK_NULL(pWrapper, code, lino, _end, terrno);
    }

    if (pWrapper != NULL && pWrapper->version != pSubmitTbData->sver) {
      // reset outdated schema
      tDeleteSchemaWrapper(pWrapper);
      pWrapper = NULL;
      taosLRUCacheUpdate(pScanInfo->pPhyTblSchemaCache, h, pWrapper);
    }

    if (pWrapper == NULL) {
      // get physical table schema from meta
      pWrapper = metaGetTableSchema(pReader->pVnodeMeta, pTbUid, pSubmitTbData->sver, 1, NULL);
      if (pWrapper == NULL) {
        tqWarn("vgId:%d, cannot found schema wrapper for table: suid:%" PRId64 ", uid:%" PRId64
               "version %d, possibly dropped table",
               pReader->pWalReader->pWal->cfg.vgId, pSubmitTbData->suid, pTbUid, pSubmitTbData->sver);
        TSDB_CHECK_NULL(pWrapper, code, lino, _end, TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND);
      }
      if (h == NULL) {
        // insert schema to cache
        code = taosLRUCacheInsert(pScanInfo->pPhyTblSchemaCache, &cacheKey, sizeof(int64_t), pWrapper, POINTER_BYTES,
                                  freeTableSchemaCache, NULL, NULL, TAOS_LRU_PRIORITY_LOW, NULL);
        if (code != TSDB_CODE_SUCCESS) {
          tDeleteSchemaWrapper(pWrapper);
        }
        TSDB_CHECK_CODE(code, lino, _end);
      } else {
        // update schema in cache
        taosLRUCacheUpdate(pScanInfo->pPhyTblSchemaCache, h, pWrapper);
      }
    } else {
      pScanInfo->cacheHit++;
    }
    TSDB_CHECK_NULL(pWrapper, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);
    pPhyTblSchema = tBuildTSchema(pWrapper->pSchema, pWrapper->nCols, pWrapper->version);
    TSDB_CHECK_NULL(pPhyTblSchema, code, lino, _end, terrno);

    numOfRows = taosArrayGetSize(pSubmitTbData->aRowP);
    nInputCols = pPhyTblSchema->numOfCols;
  }

  code = blockDataEnsureCapacity(pBlock, numOfRows);
  TSDB_CHECK_CODE(code, lino, _end);

  // convert one block
  for (int32_t i = 0, j = 1; j < nOutputCols;) {
    SColumnInfoData* pOutCol = taosArrayGet(pBlock->pDataBlock, j);
    TSDB_CHECK_NULL(pOutCol, code, lino, _end, terrno);
    if (i >= nColInfos) {
      tqInfo("%s has %d column info, but vtable column %d is missing, id: %s", __func__, nColInfos, pOutCol->info.colId,
             idstr);
      colDataSetNNULL(pOutCol, 0, numOfRows);
      j++;
      continue;
    }

    SVTColInfo* pCol = taosArrayGet(pColInfos, i);
    TSDB_CHECK_NULL(pCol, code, lino, _end, terrno);
    if (pCol->vColId < pOutCol->info.colId) {
      i++;
      continue;
    } else if (pCol->vColId > pOutCol->info.colId) {
      tqInfo("%s does not find column info for vtable column %d, closest vtable column is %d, id: %s", __func__,
             pOutCol->info.colId, pCol->vColId, idstr);
      colDataSetNNULL(pOutCol, 0, numOfRows);
      j++;
      continue;
    }

    // copy data from physical table to the result block of virtual table
    if (pCol->pTbUid != pTbUid) {
      // skip this column since it is from another physical table
    } else if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
      // try to find the corresponding column data of physical table
      SColData* pColData = NULL;
      for (int32_t k = 0; k < nInputCols; ++k) {
        pColData = taosArrayGet(pSubmitTbData->aCol, k);
        TSDB_CHECK_NULL(pColData, code, lino, _end, terrno);
        if (pColData->cid == pCol->pColId) {
          break;
        }
        pColData = NULL;
      }
      if (pColData == NULL) {
        tqError("%s does not find data of physical table %" PRId64 " column %d, virtual table: %" PRId64
                " column: %d, id: %s",
                __func__, pTbUid, pCol->pColId, vTbUid, pCol->vColId, idstr);
        colDataSetNNULL(pOutCol, 0, numOfRows);
        i++;
        j++;
        continue;
      }
      SColVal colVal = {0};
      for (int32_t k = 0; k < pColData->nVal; ++k) {
        code = tColDataGetValue(pColData, k, &colVal);
        TSDB_CHECK_CODE(code, lino, _end);
        code = doSetVal(pOutCol, k, &colVal);
        TSDB_CHECK_CODE(code, lino, _end);
      }
    } else {
      SArray* pRows = pSubmitTbData->aRowP;
      TSDB_CHECK_NULL(pPhyTblSchema, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);

      SColVal colVal = {0};
      for (int32_t k = 0; k < numOfRows; ++k) {
        SRow* pRow = taosArrayGetP(pRows, k);
        TSDB_CHECK_NULL(pRow, code, lino, _end, terrno);
        for (int32_t l = 0; l < nInputCols; ++l) {
          code = tRowGet(pRow, pPhyTblSchema, l, &colVal);
          TSDB_CHECK_CODE(code, lino, _end);
          if (colVal.cid == pCol->pColId) {
            code = doSetVal(pOutCol, k, &colVal);
            TSDB_CHECK_CODE(code, lino, _end);
            break;
          } else if (colVal.cid > pCol->pColId || l == (nInputCols - 1)) {
            colDataSetNULL(pOutCol, k);
            break;
          }
        }
      }
    }

    i++;
    j++;
  }

  // enforce to fill the first ts column
  if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
    SColumnInfoData* pOutCol = taosArrayGet(pBlock->pDataBlock, 0);
    SColData*        pColData = taosArrayGet(pSubmitTbData->aCol, 0);
    TSDB_CHECK_NULL(pColData, code, lino, _end, terrno);
    SColVal colVal = {0};
    for (int32_t k = 0; k < pColData->nVal; ++k) {
      code = tColDataGetValue(pColData, k, &colVal);
      TSDB_CHECK_CODE(code, lino, _end);
      code = doSetVal(pOutCol, k, &colVal);
      TSDB_CHECK_CODE(code, lino, _end);
    }
  } else {
    SColumnInfoData* pOutCol = taosArrayGet(pBlock->pDataBlock, 0);
    SArray*          pRows = pSubmitTbData->aRowP;
    TSDB_CHECK_NULL(pPhyTblSchema, code, lino, _end, TSDB_CODE_INTERNAL_ERROR);
    SColVal colVal = {0};
    for (int32_t k = 0; k < numOfRows; ++k) {
      SRow* pRow = taosArrayGetP(pRows, k);
      TSDB_CHECK_NULL(pRow, code, lino, _end, terrno);
      code = tRowGet(pRow, pPhyTblSchema, 0, &colVal);
      TSDB_CHECK_CODE(code, lino, _end);
      code = doSetVal(pOutCol, k, &colVal);
      TSDB_CHECK_CODE(code, lino, _end);
    }
  }

  pBlock->info.rows = numOfRows;
  pBlock->info.id.uid = vTbUid;
  pBlock->info.id.groupId = pTbUid;
  pBlock->info.version = pReader->msg.ver;
  pScanInfo->nextVirtualTableIdx++;
  if (pScanInfo->nextVirtualTableIdx >= taosArrayGetSize(pRelatedVTs)) {
    pReader->nextBlk++;
    pScanInfo->nextVirtualTableIdx = -1;
  }
  tqDebug("tq reader will retrieve next vtable data block from %p, nextBlk:%d, vtbIdx:%d, id:%s", pReader->msg.msgStr,
          pReader->nextBlk, pScanInfo->nextVirtualTableIdx, idstr);

  *pRes = pBlock;

_end:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at line %d since %s, id: %s", __func__, lino, tstrerror(code), idstr);
  }
  if (h != NULL) {
    bool bRes = taosLRUCacheRelease(pScanInfo->pPhyTblSchemaCache, h, false);
    tqTrace("release LRU cache, res %d, id: %s", bRes, idstr);
  }
  if (pPhyTblSchema != NULL) {
    taosMemoryFreeClear(pPhyTblSchema);
  }
  return code;
}

bool tqNextVTableSourceBlockImpl(STqReader* pReader, const char* idstr) {
  int32_t            code = TSDB_CODE_SUCCESS;
  int32_t            lino = 0;
  SVTSourceScanInfo* pScanInfo = NULL;

  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);

  pScanInfo = &pReader->vtSourceScanInfo;
  if (pReader->msg.msgStr == NULL || taosHashGetSize(pScanInfo->pPhysicalTables) == 0) {
    return false;
  }

  if (pScanInfo->nextVirtualTableIdx >= 0) {
    // The data still needs to be converted into the virtual table result block
    return 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, terrno);
    int64_t pTbUid = pSubmitTbData->uid;
    void*   px = taosHashGet(pScanInfo->pPhysicalTables, &pTbUid, sizeof(int64_t));
    if (px != NULL) {
      SArray* pRelatedVTs = *(SArray**)px;
      if (taosArrayGetSize(pRelatedVTs) > 0) {
        pScanInfo->nextVirtualTableIdx = 0;
        return true;
      }
    }
    tqTrace("iterator data block in hash jump block, progress:%d/%d, uid:%" PRId64 "", pReader->nextBlk, blockSz,
            pTbUid);
    pReader->nextBlk++;
  }

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

_end:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at line %d since %s, id: %s", __func__, lino, tstrerror(code), idstr);
  }
  return (code == TSDB_CODE_SUCCESS);
}

bool tqReaderIsQueriedSourceTable(STqReader* pReader, uint64_t uid) {
  if (pReader == NULL) {
    return false;
  }
  return taosHashGet(pReader->vtSourceScanInfo.pPhysicalTables, &uid, sizeof(uint64_t)) != NULL;
}

taosdata / TDengine / #3660

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