tqRead.c 36.5 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

H
Hongze Cheng 已提交
16
#include "tq.h"
L
Liu Jicong 已提交
17

L
Liu Jicong 已提交
18 19
bool isValValidForTable(STqHandle* pHandle, SWalCont* pHead) {
  if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__TABLE) {
20 21 22
    return true;
  }

L
Liu Jicong 已提交
23 24 25
  int16_t msgType = pHead->msgType;
  char*   body = pHead->body;
  int32_t bodyLen = pHead->bodyLen;
26

L
Liu Jicong 已提交
27 28 29 30 31
  int64_t  tbSuid = pHandle->execHandle.execTb.suid;
  int64_t  realTbSuid = 0;
  SDecoder coder;
  void*    data = POINTER_SHIFT(body, sizeof(SMsgHead));
  int32_t  len = bodyLen - sizeof(SMsgHead);
32 33 34 35 36 37 38 39 40 41 42 43 44
  tDecoderInit(&coder, data, len);

  if (msgType == TDMT_VND_CREATE_STB || msgType == TDMT_VND_ALTER_STB) {
    SVCreateStbReq req = {0};
    if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
      goto end;
    }
    realTbSuid = req.suid;
  } else if (msgType == TDMT_VND_DROP_STB) {
    SVDropStbReq req = {0};
    if (tDecodeSVDropStbReq(&coder, &req) < 0) {
      goto end;
    }
L
Liu Jicong 已提交
45
    realTbSuid = req.suid;
46 47 48 49 50 51
  } else if (msgType == TDMT_VND_CREATE_TABLE) {
    SVCreateTbBatchReq req = {0};
    if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) {
      goto end;
    }

L
Liu Jicong 已提交
52
    int32_t        needRebuild = 0;
53 54 55
    SVCreateTbReq* pCreateReq = NULL;
    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
      pCreateReq = req.pReqs + iReq;
L
Liu Jicong 已提交
56
      if (pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid) {
57 58 59
        needRebuild++;
      }
    }
L
Liu Jicong 已提交
60
    if (needRebuild == 0) {
61
      // do nothing
L
Liu Jicong 已提交
62
    } else if (needRebuild == req.nReqs) {
63
      realTbSuid = tbSuid;
L
Liu Jicong 已提交
64
    } else {
65 66 67 68 69
      realTbSuid = tbSuid;
      SVCreateTbBatchReq reqNew = {0};
      reqNew.pArray = taosArrayInit(req.nReqs, sizeof(struct SVCreateTbReq));
      for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
        pCreateReq = req.pReqs + iReq;
L
Liu Jicong 已提交
70
        if (pCreateReq->type == TSDB_CHILD_TABLE && pCreateReq->ctb.suid == tbSuid) {
71 72 73 74 75
          reqNew.nReqs++;
          taosArrayPush(reqNew.pArray, pCreateReq);
        }
      }

L
Liu Jicong 已提交
76
      int     tlen;
77 78 79 80 81
      int32_t ret = 0;
      tEncodeSize(tEncodeSVCreateTbBatchReq, &reqNew, tlen, ret);
      void* buf = taosMemoryMalloc(tlen);
      if (NULL == buf) {
        taosArrayDestroy(reqNew.pArray);
82 83 84 85 86 87 88
        for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
          pCreateReq = req.pReqs + iReq;
          taosMemoryFreeClear(pCreateReq->comment);
          if (pCreateReq->type == TSDB_CHILD_TABLE) {
            taosArrayDestroy(pCreateReq->ctb.tagName);
          }
        }
89 90 91 92 93 94 95 96 97 98 99
        goto end;
      }
      SEncoder coderNew = {0};
      tEncoderInit(&coderNew, buf, tlen - sizeof(SMsgHead));
      tEncodeSVCreateTbBatchReq(&coderNew, &reqNew);
      tEncoderClear(&coderNew);
      memcpy(pHead->body + sizeof(SMsgHead), buf, tlen);
      pHead->bodyLen = tlen + sizeof(SMsgHead);
      taosMemoryFree(buf);
      taosArrayDestroy(reqNew.pArray);
    }
100 101 102 103 104 105 106 107

    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
      pCreateReq = req.pReqs + iReq;
      taosMemoryFreeClear(pCreateReq->comment);
      if (pCreateReq->type == TSDB_CHILD_TABLE) {
        taosArrayDestroy(pCreateReq->ctb.tagName);
      }
    }
108
  } else if (msgType == TDMT_VND_ALTER_TABLE) {
L
Liu Jicong 已提交
109
    SVAlterTbReq req = {0};
110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130

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

    SMetaReader mr = {0};
    metaReaderInit(&mr, pHandle->execHandle.pExecReader->pVnodeMeta, 0);

    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(&coder, &req) < 0) {
      goto end;
    }

L
Liu Jicong 已提交
131
    int32_t      needRebuild = 0;
132 133 134 135
    SVDropTbReq* pDropReq = NULL;
    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
      pDropReq = req.pReqs + iReq;

L
Liu Jicong 已提交
136
      if (pDropReq->suid == tbSuid) {
137 138 139
        needRebuild++;
      }
    }
L
Liu Jicong 已提交
140
    if (needRebuild == 0) {
141
      // do nothing
L
Liu Jicong 已提交
142
    } else if (needRebuild == req.nReqs) {
143
      realTbSuid = tbSuid;
L
Liu Jicong 已提交
144
    } else {
145 146 147 148 149
      realTbSuid = tbSuid;
      SVDropTbBatchReq reqNew = {0};
      reqNew.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbReq));
      for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
        pDropReq = req.pReqs + iReq;
L
Liu Jicong 已提交
150
        if (pDropReq->suid == tbSuid) {
151 152 153 154 155
          reqNew.nReqs++;
          taosArrayPush(reqNew.pArray, pDropReq);
        }
      }

L
Liu Jicong 已提交
156
      int     tlen;
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180
      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));
      tEncodeSVDropTbBatchReq(&coderNew, &reqNew);
      tEncoderClear(&coderNew);
      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(&coder, &req) < 0) {
      goto end;
    }
    realTbSuid = req.suid;
  }

L
Liu Jicong 已提交
181
end:
182 183 184 185
  tDecoderClear(&coder);
  return tbSuid == realTbSuid;
}

wmmhello's avatar
wmmhello 已提交
186
int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** ppCkHead) {
L
Liu Jicong 已提交
187
  int32_t code = 0;
wmmhello's avatar
wmmhello 已提交
188
  taosThreadMutexLock(&pHandle->pWalReader->mutex);
L
Liu Jicong 已提交
189 190 191
  int64_t offset = *fetchOffset;

  while (1) {
wmmhello's avatar
wmmhello 已提交
192 193 194
    if (walFetchHead(pHandle->pWalReader, offset, *ppCkHead) < 0) {
      tqDebug("tmq poll: consumer:%" PRId64 ", (epoch %d) vgId:%d offset %" PRId64 ", no more log to return",
              pHandle->consumerId, pHandle->epoch, TD_VID(pTq->pVnode), offset);
L
Liu Jicong 已提交
195 196 197 198 199
      *fetchOffset = offset - 1;
      code = -1;
      goto END;
    }

200 201 202
    tqDebug("vgId:%d, taosx get msg ver %" PRId64 ", type: %s", pTq->pVnode->config.vgId, offset,
            TMSG_INFO((*ppCkHead)->head.msgType));

L
Liu Jicong 已提交
203
    if ((*ppCkHead)->head.msgType == TDMT_VND_SUBMIT) {
wmmhello's avatar
wmmhello 已提交
204
      code = walFetchBody(pHandle->pWalReader, ppCkHead);
L
Liu Jicong 已提交
205 206 207 208 209 210 211 212 213 214

      if (code < 0) {
        *fetchOffset = offset;
        code = -1;
        goto END;
      }
      *fetchOffset = offset;
      code = 0;
      goto END;
    } else {
wmmhello's avatar
wmmhello 已提交
215
      if (pHandle->fetchMeta) {
L
Liu Jicong 已提交
216
        SWalCont* pHead = &((*ppCkHead)->head);
L
Liu Jicong 已提交
217
        if (IS_META_MSG(pHead->msgType)) {
wmmhello's avatar
wmmhello 已提交
218
          code = walFetchBody(pHandle->pWalReader, ppCkHead);
L
Liu Jicong 已提交
219 220 221 222 223
          if (code < 0) {
            *fetchOffset = offset;
            code = -1;
            goto END;
          }
L
Liu Jicong 已提交
224

L
Liu Jicong 已提交
225
          if (isValValidForTable(pHandle, pHead)) {
226 227 228
            *fetchOffset = offset;
            code = 0;
            goto END;
L
Liu Jicong 已提交
229 230 231
          } else {
            offset++;
            continue;
232
          }
L
Liu Jicong 已提交
233 234
        }
      }
wmmhello's avatar
wmmhello 已提交
235
      code = walSkipFetchBody(pHandle->pWalReader, *ppCkHead);
L
Liu Jicong 已提交
236 237 238 239 240 241 242 243
      if (code < 0) {
        *fetchOffset = offset;
        code = -1;
        goto END;
      }
      offset++;
    }
  }
L
Liu Jicong 已提交
244
END:
wmmhello's avatar
wmmhello 已提交
245
  taosThreadMutexUnlock(&pHandle->pWalReader->mutex);
L
Liu Jicong 已提交
246 247 248
  return code;
}

L
Liu Jicong 已提交
249
STqReader* tqOpenReader(SVnode* pVnode) {
L
Liu Jicong 已提交
250
  STqReader* pReader = taosMemoryCalloc(1, sizeof(STqReader));
L
Liu Jicong 已提交
251
  if (pReader == NULL) {
L
Liu Jicong 已提交
252 253
    return NULL;
  }
L
Liu Jicong 已提交
254

L
Liu Jicong 已提交
255 256
  pReader->pWalReader = walOpenReader(pVnode->pWal, NULL);
  if (pReader->pWalReader == NULL) {
L
Liu Jicong 已提交
257
    taosMemoryFree(pReader);
L
Liu Jicong 已提交
258 259
    return NULL;
  }
L
Liu Jicong 已提交
260 261

  pReader->pVnodeMeta = pVnode->pMeta;
L
Liu Jicong 已提交
262
  /*pReader->pMsg = NULL;*/
L
Liu Jicong 已提交
263 264 265 266 267 268 269 270 271 272 273 274
  pReader->ver = -1;
  pReader->pColIdList = NULL;
  pReader->cachedSchemaVer = 0;
  pReader->cachedSchemaSuid = 0;
  pReader->pSchema = NULL;
  pReader->pSchemaWrapper = NULL;
  pReader->tbIdHash = NULL;
  return pReader;
}

void tqCloseReader(STqReader* pReader) {
  // close wal reader
275 276 277
  if (pReader->pWalReader) {
    walCloseReader(pReader->pWalReader);
  }
L
Liu Jicong 已提交
278
  // free cached schema
279 280 281 282 283 284 285 286 287
  if (pReader->pSchema) {
    taosMemoryFree(pReader->pSchema);
  }
  if (pReader->pSchemaWrapper) {
    tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
  }
  if (pReader->pColIdList) {
    taosArrayDestroy(pReader->pColIdList);
  }
L
Liu Jicong 已提交
288
  // free hash
289
  taosHashCleanup(pReader->tbIdHash);
L
Liu Jicong 已提交
290 291 292
  taosMemoryFree(pReader);
}

L
Liu Jicong 已提交
293
int32_t tqSeekVer(STqReader* pReader, int64_t ver) {
L
Liu Jicong 已提交
294 295 296 297
  if (walReadSeekVer(pReader->pWalReader, ver) < 0) {
    return -1;
  }
  return 0;
L
Liu Jicong 已提交
298 299
}

L
Liu Jicong 已提交
300
int32_t tqNextBlock(STqReader* pReader, SFetchRet* ret) {
L
Liu Jicong 已提交
301
  bool fromProcessedMsg = pReader->msg2.msgStr != NULL;
L
Liu Jicong 已提交
302 303 304 305

  while (1) {
    if (!fromProcessedMsg) {
      if (walNextValidMsg(pReader->pWalReader) < 0) {
306 307
        pReader->ver =
            pReader->pWalReader->curVersion - (pReader->pWalReader->curInvalid | pReader->pWalReader->curStopped);
L
Liu Jicong 已提交
308 309
        ret->offset.type = TMQ_OFFSET__LOG;
        ret->offset.version = pReader->ver;
L
Liu Jicong 已提交
310
        ret->fetchType = FETCH_TYPE__NONE;
S
Shengliang Guan 已提交
311
        tqDebug("return offset %" PRId64 ", no more valid", ret->offset.version);
L
Liu Jicong 已提交
312 313
        return -1;
      }
L
Liu Jicong 已提交
314 315
      void*   body = POINTER_SHIFT(pReader->pWalReader->pHead->head.body, sizeof(SMsgHead));
      int32_t bodyLen = pReader->pWalReader->pHead->head.bodyLen - sizeof(SMsgHead);
L
Liu Jicong 已提交
316
      int64_t ver = pReader->pWalReader->pHead->head.version;
L
Liu Jicong 已提交
317
#if 0
L
Liu Jicong 已提交
318 319 320 321 322 323
      if (pReader->pWalReader->pHead->head.msgType != TDMT_VND_SUBMIT) {
        // TODO do filter
        ret->fetchType = FETCH_TYPE__META;
        ret->meta = pReader->pWalReader->pHead->head.body;
        return 0;
      } else {
L
Liu Jicong 已提交
324
#endif
L
Liu Jicong 已提交
325 326
      tqReaderSetSubmitReq2(pReader, body, bodyLen, ver);
      /*tqReaderSetDataMsg(pReader, body, pReader->pWalReader->pHead->head.version);*/
L
Liu Jicong 已提交
327
#if 0
L
Liu Jicong 已提交
328
      }
L
Liu Jicong 已提交
329
#endif
L
Liu Jicong 已提交
330 331
    }

L
Liu Jicong 已提交
332
    while (tqNextDataBlock2(pReader)) {
L
Liu Jicong 已提交
333
      // TODO mem free
L
Liu Jicong 已提交
334
      memset(&ret->data, 0, sizeof(SSDataBlock));
335
      int32_t code = tqRetrieveDataBlock2(&ret->data, pReader, NULL);
L
Liu Jicong 已提交
336
      if (code != 0 || ret->data.info.rows == 0) {
L
Liu Jicong 已提交
337
        continue;
L
Liu Jicong 已提交
338 339
      }
      ret->fetchType = FETCH_TYPE__DATA;
L
Liu Jicong 已提交
340
      tqDebug("return data rows %d", ret->data.info.rows);
L
Liu Jicong 已提交
341 342 343 344
      return 0;
    }

    if (fromProcessedMsg) {
L
Liu Jicong 已提交
345 346
      ret->offset.type = TMQ_OFFSET__LOG;
      ret->offset.version = pReader->ver;
L
Liu Jicong 已提交
347
      ret->fetchType = FETCH_TYPE__SEP;
S
Shengliang Guan 已提交
348
      tqDebug("return offset %" PRId64 ", processed finish", ret->offset.version);
L
Liu Jicong 已提交
349 350 351
      return 0;
    }
  }
L
Liu Jicong 已提交
352 353
}

L
Liu Jicong 已提交
354
#if 0
L
Liu Jicong 已提交
355
int32_t tqReaderSetDataMsg(STqReader* pReader, const SSubmitReq* pMsg, int64_t ver) {
L
Liu Jicong 已提交
356
  pReader->pMsg = pMsg;
L
Liu Jicong 已提交
357

L
Liu Jicong 已提交
358
  if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1;
L
Liu Jicong 已提交
359
  while (true) {
L
Liu Jicong 已提交
360 361
    if (tGetSubmitMsgNext(&pReader->msgIter, &pReader->pBlock) < 0) return -1;
    if (pReader->pBlock == NULL) break;
L
Liu Jicong 已提交
362 363
  }

L
Liu Jicong 已提交
364 365 366
  if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1;
  pReader->ver = ver;
  memset(&pReader->blkIter, 0, sizeof(SSubmitBlkIter));
L
Liu Jicong 已提交
367
  return 0;
L
Liu Jicong 已提交
368
}
L
Liu Jicong 已提交
369 370 371 372
#endif

int32_t tqReaderSetSubmitReq2(STqReader* pReader, void* msgStr, int32_t msgLen, int64_t ver) {
  ASSERT(pReader->msg2.msgStr == NULL);
L
Liu Jicong 已提交
373 374
  ASSERT(msgStr);
  ASSERT(msgLen);
L
Liu Jicong 已提交
375
  ASSERT(ver >= 0);
L
Liu Jicong 已提交
376 377 378
  pReader->msg2.msgStr = msgStr;
  pReader->msg2.msgLen = msgLen;
  pReader->msg2.ver = ver;
L
Liu Jicong 已提交
379
  pReader->ver = ver;
L
Liu Jicong 已提交
380

L
Liu Jicong 已提交
381
  tqDebug("tq reader set msg %p %d", msgStr, msgLen);
L
Liu Jicong 已提交
382 383

  if (pReader->setMsg == 0) {
L
Liu Jicong 已提交
384 385
    SDecoder decoder;
    tDecoderInit(&decoder, pReader->msg2.msgStr, pReader->msg2.msgLen);
L
Liu Jicong 已提交
386 387 388
    if (tDecodeSSubmitReq2(&decoder, &pReader->submit) < 0) {
      ASSERT(0);
    }
L
Liu Jicong 已提交
389
    tDecoderClear(&decoder);
L
Liu Jicong 已提交
390
    pReader->setMsg = 1;
L
Liu Jicong 已提交
391 392 393
  }
  return 0;
}
L
Liu Jicong 已提交
394

L
Liu Jicong 已提交
395
#if 0
L
Liu Jicong 已提交
396 397
bool tqNextDataBlock(STqReader* pReader) {
  if (pReader->pMsg == NULL) return false;
L
Liu Jicong 已提交
398
  while (1) {
L
Liu Jicong 已提交
399
    if (tGetSubmitMsgNext(&pReader->msgIter, &pReader->pBlock) < 0) {
L
Liu Jicong 已提交
400 401
      return false;
    }
L
Liu Jicong 已提交
402 403
    if (pReader->pBlock == NULL) {
      pReader->pMsg = NULL;
5
54liuyao 已提交
404 405
      return false;
    }
L
Liu Jicong 已提交
406

L
Liu Jicong 已提交
407
    if (pReader->tbIdHash == NULL) {
L
Liu Jicong 已提交
408 409
      return true;
    }
L
Liu Jicong 已提交
410
    void* ret = taosHashGet(pReader->tbIdHash, &pReader->msgIter.uid, sizeof(int64_t));
S
Shengliang Guan 已提交
411
    /*tqDebug("search uid %" PRId64, pHandle->msgIter.uid);*/
L
Liu Jicong 已提交
412
    if (ret != NULL) {
S
Shengliang Guan 已提交
413
      /*tqDebug("find   uid %" PRId64, pHandle->msgIter.uid);*/
L
Liu Jicong 已提交
414
      return true;
415 416 417 418
    }
  }
  return false;
}
L
Liu Jicong 已提交
419 420 421 422
#endif

bool tqNextDataBlock2(STqReader* pReader) {
  if (pReader->msg2.msgStr == NULL) return false;
L
Liu Jicong 已提交
423
  ASSERT(pReader->setMsg == 1);
L
Liu Jicong 已提交
424

L
Liu Jicong 已提交
425 426 427 428
  tqDebug("tq reader next data block %p, %d %" PRId64 " %d", pReader->msg2.msgStr, pReader->msg2.msgLen,
          pReader->msg2.ver, pReader->nextBlk);

  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
L
Liu Jicong 已提交
429
  while (pReader->nextBlk < blockSz) {
L
Liu Jicong 已提交
430
    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
L
Liu Jicong 已提交
431 432
    ASSERT(pSubmitTbData->uid);

L
Liu Jicong 已提交
433 434 435 436 437 438
    if (pReader->tbIdHash == NULL) return true;

    void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t));
    if (ret != NULL) {
      return true;
    }
L
Liu Jicong 已提交
439
    pReader->nextBlk++;
L
Liu Jicong 已提交
440 441
  }

L
Liu Jicong 已提交
442 443
  tDestroySSubmitReq2(&pReader->submit, TSDB_MSG_FLG_DECODE);
  pReader->setMsg = 0;
L
Liu Jicong 已提交
444 445 446 447 448 449 450 451
  pReader->nextBlk = 0;
  pReader->msg2.msgStr = NULL;

  return false;
}

bool tqNextDataBlockFilterOut2(STqReader* pReader, SHashObj* filterOutUids) {
  if (pReader->msg2.msgStr == NULL) return false;
L
Liu Jicong 已提交
452
  ASSERT(pReader->setMsg == 1);
L
Liu Jicong 已提交
453

L
Liu Jicong 已提交
454
  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
L
Liu Jicong 已提交
455
  while (pReader->nextBlk < blockSz) {
L
Liu Jicong 已提交
456
    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
L
Liu Jicong 已提交
457 458 459 460 461 462 463 464
    if (pReader->tbIdHash == NULL) return true;

    void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t));
    if (ret == NULL) {
      return true;
    }
  }

L
Liu Jicong 已提交
465 466
  tDestroySSubmitReq2(&pReader->submit, TSDB_MSG_FLG_DECODE);
  pReader->setMsg = 0;
L
Liu Jicong 已提交
467 468 469 470 471
  pReader->nextBlk = 0;
  pReader->msg2.msgStr = NULL;

  return false;
}
472

473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501
int32_t tqMaskBlock(SSchemaWrapper* pDst, SSDataBlock* pBlock, const SSchemaWrapper* pSrc, char* mask) {
  int32_t code;

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

  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);
      code = blockDataAppendColInfo(pBlock, &colInfo);
      if (code != 0) {
        return -1;
      }
    }
  }
  return 0;
}

L
Liu Jicong 已提交
502
#if 0
L
Liu Jicong 已提交
503
bool tqNextDataBlockFilterOut(STqReader* pHandle, SHashObj* filterOutUids) {
504 505 506 507 508 509 510 511 512
  while (1) {
    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
      return false;
    }
    if (pHandle->pBlock == NULL) return false;

    void* ret = taosHashGet(filterOutUids, &pHandle->msgIter.uid, sizeof(int64_t));
    if (ret == NULL) {
      return true;
L
Liu Jicong 已提交
513 514 515 516 517
    }
  }
  return false;
}

L
Liu Jicong 已提交
518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552
int32_t tqScanSubmitSplit(SArray* pBlocks, SArray* schemas, STqReader* pReader) {
  //
  int32_t sversion = htonl(pReader->pBlock->sversion);
  if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion ||
      pReader->cachedSchemaSuid != pReader->msgIter.suid) {
    taosMemoryFree(pReader->pSchema);
    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
    if (pReader->pSchema == NULL) {
      tqWarn("vgId:%d, cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64
             "), version %d, possibly dropped table",
             pReader->pWalReader->pWal->cfg.vgId, pReader->msgIter.uid, pReader->msgIter.suid, sversion);
      pReader->cachedSchemaSuid = 0;
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }

    tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
    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, pReader->msgIter.uid, pReader->cachedSchemaVer);
      pReader->cachedSchemaSuid = 0;
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }

    STSchema*       pTschema = pReader->pSchema;
    SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;

    int32_t colNumNeed = taosArrayGetSize(pReader->pColIdList);
  }
  return 0;
}
#endif

553
int32_t tqRetrieveDataBlock2(SSDataBlock* pBlock, STqReader* pReader, SSubmitTbData** pSubmitTbDataRet) {
L
Liu Jicong 已提交
554
  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
L
Liu Jicong 已提交
555 556
  ASSERT(pReader->nextBlk < blockSz);

L
Liu Jicong 已提交
557 558 559
  tqDebug("tq reader retrieve data block %p, %d", pReader->msg2.msgStr, pReader->nextBlk);

  SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
L
Liu Jicong 已提交
560 561
  pReader->nextBlk++;

562
  if(pSubmitTbDataRet) *pSubmitTbDataRet = pSubmitTbData;
L
Liu Jicong 已提交
563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673
  int32_t sversion = pSubmitTbData->sver;
  int64_t suid = pSubmitTbData->suid;
  int64_t uid = pSubmitTbData->uid;
  pReader->lastBlkUid = uid;

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

  if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion || pReader->cachedSchemaSuid != suid) {
    taosMemoryFree(pReader->pSchema);
    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, uid, sversion, 1);
    if (pReader->pSchema == NULL) {
      tqWarn("vgId:%d, cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64
             "), version %d, possibly dropped table",
             pReader->pWalReader->pWal->cfg.vgId, uid, suid, sversion);
      pReader->cachedSchemaSuid = 0;
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }

    tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, uid, sversion, 1);
    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;
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }

    STSchema*       pTschema = pReader->pSchema;
    SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;

    int32_t colNumNeed = taosArrayGetSize(pReader->pColIdList);

    if (colNumNeed == 0) {
      int32_t colMeta = 0;
      while (colMeta < pSchemaWrapper->nCols) {
        SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
        SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
        int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
        if (code != TSDB_CODE_SUCCESS) {
          goto FAIL;
        }
        colMeta++;
      }
    } else {
      if (colNumNeed > pSchemaWrapper->nCols) {
        colNumNeed = pSchemaWrapper->nCols;
      }

      int32_t colMeta = 0;
      int32_t colNeed = 0;
      while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
        SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
        col_id_t colIdSchema = pColSchema->colId;
        col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, colNeed);
        if (colIdSchema < colIdNeed) {
          colMeta++;
        } else if (colIdSchema > colIdNeed) {
          colNeed++;
        } else {
          SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
          int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
          if (code != TSDB_CODE_SUCCESS) {
            goto FAIL;
          }
          colMeta++;
          colNeed++;
        }
      }
    }

    int32_t numOfRows = 0;

    if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {
      SArray*   pCols = pSubmitTbData->aCol;
      SColData* pCol = taosArrayGet(pCols, 0);
      numOfRows = pCol->nVal;
    } else {
      SArray* pRows = pSubmitTbData->aRowP;
      numOfRows = taosArrayGetSize(pRows);
    }

    if (blockDataEnsureCapacity(pBlock, numOfRows) < 0) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto FAIL;
    }
    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) {
        ASSERT(sourceIdx < numOfCols);

        SColData*        pCol = taosArrayGet(pCols, sourceIdx);
        SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, targetIdx);
        SColVal          colVal;

        ASSERT(pCol->nVal == numOfRows);

        if (pCol->cid < pColData->info.colId) {
          sourceIdx++;
        } else if (pCol->cid == pColData->info.colId) {
          for (int32_t i = 0; i < pCol->nVal; i++) {
wmmhello's avatar
wmmhello 已提交
674
            tColDataGetValue(pCol, i, &colVal);
L
Liu Jicong 已提交
675
#if 0
L
Liu Jicong 已提交
676 677 678 679 680 681
            void* val = NULL;
            if (IS_STR_DATA_TYPE(colVal.type)) {
              val = colVal.value.pData;
            } else {
              val = &colVal.value.val;
            }
L
Liu Jicong 已提交
682
            if (colDataAppend(pColData, i, val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
L
Liu Jicong 已提交
683 684
              goto FAIL;
            }
L
Liu Jicong 已提交
685 686
#endif
            if (IS_STR_DATA_TYPE(colVal.type)) {
L
Liu Jicong 已提交
687 688 689 690 691 692 693 694 695
              if (colVal.value.pData != NULL) {
                char val[65535 + 2];
                memcpy(varDataVal(val), colVal.value.pData, colVal.value.nData);
                varDataSetLen(val, colVal.value.nData);
                if (colDataAppend(pColData, i, val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
                  goto FAIL;
                }
              } else {
                colDataAppendNULL(pColData, i);
L
Liu Jicong 已提交
696 697 698 699 700 701
              }
            } else {
              if (colDataAppend(pColData, i, (void*)&colVal.value.val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
                goto FAIL;
              }
            }
L
Liu Jicong 已提交
702 703 704 705 706 707 708 709 710 711 712
          }
          sourceIdx++;
          targetIdx++;
        } else {
          ASSERT(0);
        }
      }
    } else {
      SArray* pRows = pSubmitTbData->aRowP;

      for (int32_t i = 0; i < numOfRows; i++) {
L
Liu Jicong 已提交
713
        SRow*   pRow = taosArrayGetP(pRows, i);
L
Liu Jicong 已提交
714 715 716 717 718
        int32_t sourceIdx = 0;

        for (int32_t j = 0; j < colActual; j++) {
          SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, j);
          while (1) {
L
Liu Jicong 已提交
719
            ASSERT(sourceIdx < pTschema->numOfCols);
L
Liu Jicong 已提交
720 721 722 723 724 725 726 727

            SColVal colVal;
            tRowGet(pRow, pTschema, sourceIdx, &colVal);
            if (colVal.cid < pColData->info.colId) {
              sourceIdx++;
              continue;
            } else if (colVal.cid == pColData->info.colId) {
              if (IS_STR_DATA_TYPE(colVal.type)) {
L
Liu Jicong 已提交
728 729 730 731 732 733 734 735 736
                if (colVal.value.pData != NULL) {
                  char val[65535 + 2];
                  memcpy(varDataVal(val), colVal.value.pData, colVal.value.nData);
                  varDataSetLen(val, colVal.value.nData);
                  if (colDataAppend(pColData, i, val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
                    goto FAIL;
                  }
                } else {
                  colDataAppendNULL(pColData, i);
L
Liu Jicong 已提交
737 738
                }
                /*val = colVal.value.pData;*/
L
Liu Jicong 已提交
739
              } else {
L
Liu Jicong 已提交
740 741 742
                if (colDataAppend(pColData, i, (void*)&colVal.value.val, !COL_VAL_IS_VALUE(&colVal)) < 0) {
                  goto FAIL;
                }
L
Liu Jicong 已提交
743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763
              }

              sourceIdx++;
              break;
            } else {
              ASSERT(0);
            }
          }
        }
      }
    }
  }

  return 0;

FAIL:
  blockDataFreeRes(pBlock);
  return -1;
}

#if 0
L
Liu Jicong 已提交
764
int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader) {
765
  // TODO: cache multiple schema
L
Liu Jicong 已提交
766 767 768 769
  int32_t sversion = htonl(pReader->pBlock->sversion);
  if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion ||
      pReader->cachedSchemaSuid != pReader->msgIter.suid) {
    if (pReader->pSchema) taosMemoryFree(pReader->pSchema);
770
    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
L
Liu Jicong 已提交
771
    if (pReader->pSchema == NULL) {
S
Shengliang Guan 已提交
772
      tqWarn("cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64 "), version %d, possibly dropped table",
L
Liu Jicong 已提交
773
             pReader->msgIter.uid, pReader->msgIter.suid, pReader->cachedSchemaVer);
774
      pReader->cachedSchemaSuid = 0;
775
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
L
Liu Jicong 已提交
776 777
      return -1;
    }
L
Liu Jicong 已提交
778

L
Liu Jicong 已提交
779
    if (pReader->pSchemaWrapper) tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
780
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
L
Liu Jicong 已提交
781
    if (pReader->pSchemaWrapper == NULL) {
S
Shengliang Guan 已提交
782
      tqWarn("cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
L
Liu Jicong 已提交
783
             pReader->msgIter.uid, pReader->cachedSchemaVer);
784
      pReader->cachedSchemaSuid = 0;
L
Liu Jicong 已提交
785 786 787
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }
L
Liu Jicong 已提交
788 789
    pReader->cachedSchemaVer = sversion;
    pReader->cachedSchemaSuid = pReader->msgIter.suid;
L
Liu Jicong 已提交
790 791
  }

L
Liu Jicong 已提交
792 793
  STSchema*       pTschema = pReader->pSchema;
  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
L
Liu Jicong 已提交
794

L
Liu Jicong 已提交
795
  int32_t colNumNeed = taosArrayGetSize(pReader->pColIdList);
L
Liu Jicong 已提交
796

L
Liu Jicong 已提交
797 798 799 800
  if (colNumNeed == 0) {
    int32_t colMeta = 0;
    while (colMeta < pSchemaWrapper->nCols) {
      SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
801
      SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
L
Liu Jicong 已提交
802
      int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
803
      if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
804
        goto FAIL;
L
Liu Jicong 已提交
805
      }
L
Liu Jicong 已提交
806
      colMeta++;
L
Liu Jicong 已提交
807 808 809 810 811 812 813 814 815 816 817
    }
  } else {
    if (colNumNeed > pSchemaWrapper->nCols) {
      colNumNeed = pSchemaWrapper->nCols;
    }

    int32_t colMeta = 0;
    int32_t colNeed = 0;
    while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
      SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
      col_id_t colIdSchema = pColSchema->colId;
L
Liu Jicong 已提交
818
      col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, colNeed);
L
Liu Jicong 已提交
819 820 821 822 823
      if (colIdSchema < colIdNeed) {
        colMeta++;
      } else if (colIdSchema > colIdNeed) {
        colNeed++;
      } else {
824
        SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
L
Liu Jicong 已提交
825
        int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
826
        if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
827 828 829 830 831
          goto FAIL;
        }
        colMeta++;
        colNeed++;
      }
L
Liu Jicong 已提交
832 833
    }
  }
L
Liu Jicong 已提交
834

L
Liu Jicong 已提交
835
  if (blockDataEnsureCapacity(pBlock, pReader->msgIter.numOfRows) < 0) {
836
    terrno = TSDB_CODE_OUT_OF_MEMORY;
837 838 839 840
    goto FAIL;
  }

  int32_t colActual = blockDataGetNumOfCols(pBlock);
L
Liu Jicong 已提交
841

L
Liu Jicong 已提交
842 843 844 845
  STSRowIter iter = {0};
  tdSTSRowIterInit(&iter, pTschema);
  STSRow* row;
  int32_t curRow = 0;
846

L
Liu Jicong 已提交
847
  tInitSubmitBlkIter(&pReader->msgIter, pReader->pBlock, &pReader->blkIter);
848

H
Haojun Liao 已提交
849
  pBlock->info.id.uid = pReader->msgIter.uid;
L
Liu Jicong 已提交
850
  pBlock->info.rows = pReader->msgIter.numOfRows;
851
  pBlock->info.version = pReader->pMsg->version;
852
  pBlock->info.dataLoad = 1;
853

L
Liu Jicong 已提交
854
  while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) {
L
Liu Jicong 已提交
855 856
    tdSTSRowIterReset(&iter, row);
    // get all wanted col of that block
L
Liu Jicong 已提交
857
    for (int32_t i = 0; i < colActual; i++) {
858
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
L
Liu Jicong 已提交
859
      SCellVal         sVal = {0};
C
Cary Xu 已提交
860
      if (!tdSTSRowIterFetch(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
L
Liu Jicong 已提交
861 862
        break;
      }
863
      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType != TD_VTYPE_NORM) < 0) {
L
Liu Jicong 已提交
864
        goto FAIL;
L
Liu Jicong 已提交
865
      }
L
Liu Jicong 已提交
866 867 868
    }
    curRow++;
  }
L
Liu Jicong 已提交
869
  return 0;
870

L
Liu Jicong 已提交
871
FAIL:
872
  blockDataFreeRes(pBlock);
L
Liu Jicong 已提交
873
  return -1;
L
Liu Jicong 已提交
874
}
H
Hongze Cheng 已提交
875

876
int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas) {
877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909
  int32_t sversion = htonl(pReader->pBlock->sversion);

  if (pReader->cachedSchemaSuid == 0 || pReader->cachedSchemaVer != sversion ||
      pReader->cachedSchemaSuid != pReader->msgIter.suid) {
    if (pReader->pSchema) taosMemoryFree(pReader->pSchema);
    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
    if (pReader->pSchema == NULL) {
      tqWarn("cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64 "), version %d, possibly dropped table",
             pReader->msgIter.uid, pReader->msgIter.suid, pReader->cachedSchemaVer);
      pReader->cachedSchemaSuid = 0;
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }

    if (pReader->pSchemaWrapper) tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
    if (pReader->pSchemaWrapper == NULL) {
      tqWarn("cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
             pReader->msgIter.uid, pReader->cachedSchemaVer);
      pReader->cachedSchemaSuid = 0;
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }
    pReader->cachedSchemaVer = sversion;
    pReader->cachedSchemaSuid = pReader->msgIter.suid;
  }

  STSchema*       pTschema = pReader->pSchema;
  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;

  int32_t colAtMost = pSchemaWrapper->nCols;

  int32_t curRow = 0;
910
  int32_t lastRow = 0;
911 912

  char* assigned = taosMemoryCalloc(1, pSchemaWrapper->nCols);
913
  if (assigned == NULL) return -1;
914 915 916 917 918 919 920 921 922 923

  tInitSubmitBlkIter(&pReader->msgIter, pReader->pBlock, &pReader->blkIter);
  STSRowIter iter = {0};
  tdSTSRowIterInit(&iter, pTschema);
  STSRow* row;

  while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) {
    bool buildNew = false;
    tdSTSRowIterReset(&iter, row);

924
    tqDebug("vgId:%d, row of block %d", pReader->pWalReader->pWal->cfg.vgId, curRow);
925 926 927 928 929
    for (int32_t i = 0; i < colAtMost; i++) {
      SCellVal sVal = {0};
      if (!tdSTSRowIterFetch(&iter, pSchemaWrapper->pSchema[i].colId, pSchemaWrapper->pSchema[i].type, &sVal)) {
        break;
      }
930
      tqDebug("vgId:%d, %d col, type %d", pReader->pWalReader->pWal->cfg.vgId, i, sVal.valType);
931 932 933 934 935 936 937 938 939 940 941 942 943
      if (curRow == 0) {
        assigned[i] = sVal.valType != TD_VTYPE_NONE;
        buildNew = true;
      } else {
        bool currentRowAssigned = sVal.valType != TD_VTYPE_NONE;
        if (currentRowAssigned != assigned[i]) {
          assigned[i] = currentRowAssigned;
          buildNew = true;
        }
      }
    }

    if (buildNew) {
944 945 946 947 948 949 950 951 952
      if (taosArrayGetSize(blocks) > 0) {
        SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
        pLastBlock->info.rows = curRow - lastRow;
        lastRow = curRow;
      }
      SSDataBlock*    pBlock = createDataBlock();
      SSchemaWrapper* pSW = taosMemoryCalloc(1, sizeof(SSchemaWrapper));
      if (tqMaskBlock(pSW, pBlock, pSchemaWrapper, assigned) < 0) {
        blockDataDestroy(pBlock);
953 954
        goto FAIL;
      }
955 956 957 958 959 960
      SSDataBlock block = {0};
      assignOneDataBlock(&block, pBlock);
      blockDataDestroy(pBlock);

      tqDebug("vgId:%d, build new block, col %d", pReader->pWalReader->pWal->cfg.vgId,
              (int32_t)taosArrayGetSize(block.pDataBlock));
961 962

      taosArrayPush(blocks, &block);
963
      taosArrayPush(schemas, &pSW);
964 965 966
    }

    SSDataBlock* pBlock = taosArrayGetLast(blocks);
H
Haojun Liao 已提交
967
    pBlock->info.id.uid = pReader->msgIter.uid;
968
    pBlock->info.rows = 0;
969 970
    pBlock->info.version = pReader->pMsg->version;

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

974 975 976 977 978
    if (blockDataEnsureCapacity(pBlock, pReader->msgIter.numOfRows - curRow) < 0) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto FAIL;
    }

979
    tdSTSRowIterReset(&iter, row);
980 981 982 983 984 985 986 987
    for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); i++) {
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
      SCellVal         sVal = {0};

      if (!tdSTSRowIterFetch(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
        break;
      }

988
      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
989 990
        goto FAIL;
      }
991 992
      tqDebug("vgId:%d, row %d col %d append %d", pReader->pWalReader->pWal->cfg.vgId, curRow, i,
              sVal.valType == TD_VTYPE_NULL);
993 994 995
    }
    curRow++;
  }
996 997
  SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
  pLastBlock->info.rows = curRow - lastRow;
998 999 1000 1001 1002 1003 1004 1005

  taosMemoryFree(assigned);
  return 0;

FAIL:
  taosMemoryFree(assigned);
  return -1;
}
L
Liu Jicong 已提交
1006 1007
#endif

1008
int32_t tqRetrieveTaosxBlock2(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) {
L
Liu Jicong 已提交
1009
  SSDataBlock block = {0};
1010
  if (tqRetrieveDataBlock2(&block, pReader, pSubmitTbDataRet) == 0) {
L
Liu Jicong 已提交
1011 1012 1013 1014 1015 1016 1017
    taosArrayPush(blocks, &block);
    SSchemaWrapper* pSW = tCloneSSchemaWrapper(pReader->pSchemaWrapper);
    taosArrayPush(schemas, &pSW);
    return 0;
  }
  return -1;
}
1018

L
Liu Jicong 已提交
1019
void tqReaderSetColIdList(STqReader* pReader, SArray* pColIdList) { pReader->pColIdList = pColIdList; }
H
Hongze Cheng 已提交
1020

L
Liu Jicong 已提交
1021 1022 1023 1024
int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash) {
    taosHashClear(pReader->tbIdHash);
  } else {
5
54liuyao 已提交
1025
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
H
Hongze Cheng 已提交
1026 1027
  }

L
Liu Jicong 已提交
1028
  if (pReader->tbIdHash == NULL) {
H
Hongze Cheng 已提交
1029 1030 1031 1032 1033 1034
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  for (int i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
L
Liu Jicong 已提交
1035
    taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
H
Hongze Cheng 已提交
1036 1037 1038 1039 1040
  }

  return 0;
}

L
Liu Jicong 已提交
1041 1042
int tqReaderAddTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash == NULL) {
5
54liuyao 已提交
1043
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
1044
    if (pReader->tbIdHash == NULL) {
H
Hongze Cheng 已提交
1045 1046 1047 1048 1049 1050 1051
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }
  }

  for (int i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
L
Liu Jicong 已提交
1052
    taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
H
Hongze Cheng 已提交
1053 1054 1055 1056
  }

  return 0;
}
1057

L
Liu Jicong 已提交
1058
int tqReaderRemoveTbUidList(STqReader* pReader, const SArray* tbUidList) {
L
Liu Jicong 已提交
1059 1060
  for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
L
Liu Jicong 已提交
1061
    taosHashRemove(pReader->tbIdHash, pKey, sizeof(int64_t));
1062 1063 1064 1065
  }

  return 0;
}
L
Liu Jicong 已提交
1066 1067 1068 1069

int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
  void* pIter = NULL;
  while (1) {
1070
    pIter = taosHashIterate(pTq->pHandle, pIter);
L
Liu Jicong 已提交
1071 1072 1073
    if (pIter == NULL) break;
    STqHandle* pExec = (STqHandle*)pIter;
    if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
1074
      int32_t code = qUpdateQualifiedTableId(pExec->execHandle.task, tbUidList, isAdd);
L
Liu Jicong 已提交
1075 1076 1077 1078
      if (code != 0) {
        tqError("update qualified table error for %s", pExec->subKey);
        continue;
      }
L
Liu Jicong 已提交
1079 1080 1081 1082 1083
    } else if (pExec->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);
L
Liu Jicong 已提交
1084
          taosHashPut(pExec->execHandle.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0);
L
Liu Jicong 已提交
1085 1086
        }
      }
1087 1088 1089 1090 1091 1092 1093 1094
    } else if (pExec->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
      if (isAdd) {
        SArray*     qa = taosArrayInit(4, sizeof(tb_uid_t));
        SMetaReader mr = {0};
        metaReaderInit(&mr, pTq->pVnode->pMeta, 0);
        for (int32_t i = 0; i < taosArrayGetSize(tbUidList); ++i) {
          uint64_t* id = (uint64_t*)taosArrayGet(tbUidList, i);

H
Haojun Liao 已提交
1095
          int32_t code = metaGetTableEntryByUidCache(&mr, *id);
1096
          if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
1097
            tqError("failed to get table meta, uid:%" PRIu64 " code:%s", *id, tstrerror(terrno));
1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117
            continue;
          }

          tDecoderClear(&mr.coder);

          if (mr.me.type != TSDB_CHILD_TABLE || mr.me.ctbEntry.suid != pExec->execHandle.execTb.suid) {
            tqDebug("table uid %" PRId64 " does not add to tq handle", *id);
            continue;
          }
          tqDebug("table uid %" PRId64 " add to tq handle", *id);
          taosArrayPush(qa, id);
        }
        metaReaderClear(&mr);
        if (taosArrayGetSize(qa) > 0) {
          tqReaderAddTbUidList(pExec->execHandle.pExecReader, qa);
        }
        taosArrayDestroy(qa);
      } else {
        // TODO handle delete table from stb
      }
L
Liu Jicong 已提交
1118 1119 1120
    }
  }
  while (1) {
L
Liu Jicong 已提交
1121
    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
L
Liu Jicong 已提交
1122
    if (pIter == NULL) break;
1123
    SStreamTask* pTask = *(SStreamTask**)pIter;
1124
    if (pTask->taskLevel == TASK_LEVEL__SOURCE) {
L
Liu Jicong 已提交
1125
      int32_t code = qUpdateQualifiedTableId(pTask->exec.executor, tbUidList, isAdd);
L
Liu Jicong 已提交
1126 1127 1128 1129
      if (code != 0) {
        tqError("update qualified table error for stream task %d", pTask->taskId);
        continue;
      }
L
Liu Jicong 已提交
1130 1131 1132 1133
    }
  }
  return 0;
}