tqRead.c 19.7 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
      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 已提交
179
  } else {
180 181 182
    ASSERT(0);
  }

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

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

  while (1) {
wmmhello's avatar
wmmhello 已提交
194 195 196
    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 已提交
197 198 199 200 201
      *fetchOffset = offset - 1;
      code = -1;
      goto END;
    }

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

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

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

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

L
Liu Jicong 已提交
254 255 256
STqReader* tqOpenReader(SVnode* pVnode) {
  STqReader* pReader = taosMemoryMalloc(sizeof(STqReader));
  if (pReader == NULL) {
L
Liu Jicong 已提交
257 258
    return NULL;
  }
L
Liu Jicong 已提交
259

L
Liu Jicong 已提交
260 261
  pReader->pWalReader = walOpenReader(pVnode->pWal, NULL);
  if (pReader->pWalReader == NULL) {
L
Liu Jicong 已提交
262
    taosMemoryFree(pReader);
L
Liu Jicong 已提交
263 264
    return NULL;
  }
L
Liu Jicong 已提交
265 266 267 268 269 270 271 272 273 274 275 276 277 278 279

  pReader->pVnodeMeta = pVnode->pMeta;
  pReader->pMsg = NULL;
  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
280 281 282
  if (pReader->pWalReader) {
    walCloseReader(pReader->pWalReader);
  }
L
Liu Jicong 已提交
283
  // free cached schema
284 285 286 287 288 289 290 291 292
  if (pReader->pSchema) {
    taosMemoryFree(pReader->pSchema);
  }
  if (pReader->pSchemaWrapper) {
    tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
  }
  if (pReader->pColIdList) {
    taosArrayDestroy(pReader->pColIdList);
  }
L
Liu Jicong 已提交
293
  // free hash
294
  taosHashCleanup(pReader->tbIdHash);
L
Liu Jicong 已提交
295 296 297
  taosMemoryFree(pReader);
}

L
Liu Jicong 已提交
298
int32_t tqSeekVer(STqReader* pReader, int64_t ver) {
L
Liu Jicong 已提交
299 300 301 302 303 304 305
  if (walReadSeekVer(pReader->pWalReader, ver) < 0) {
    ASSERT(pReader->pWalReader->curInvalid);
    ASSERT(pReader->pWalReader->curVersion == ver);
    return -1;
  }
  ASSERT(pReader->pWalReader->curVersion == ver);
  return 0;
L
Liu Jicong 已提交
306 307
}

L
Liu Jicong 已提交
308 309 310 311 312 313
int32_t tqNextBlock(STqReader* pReader, SFetchRet* ret) {
  bool fromProcessedMsg = pReader->pMsg != NULL;

  while (1) {
    if (!fromProcessedMsg) {
      if (walNextValidMsg(pReader->pWalReader) < 0) {
314 315
        pReader->ver =
            pReader->pWalReader->curVersion - (pReader->pWalReader->curInvalid | pReader->pWalReader->curStopped);
L
Liu Jicong 已提交
316 317
        ret->offset.type = TMQ_OFFSET__LOG;
        ret->offset.version = pReader->ver;
L
Liu Jicong 已提交
318
        ret->fetchType = FETCH_TYPE__NONE;
S
Shengliang Guan 已提交
319
        tqDebug("return offset %" PRId64 ", no more valid", ret->offset.version);
L
Liu Jicong 已提交
320
        ASSERT(ret->offset.version >= 0);
L
Liu Jicong 已提交
321 322 323
        return -1;
      }
      void* body = pReader->pWalReader->pHead->head.body;
L
Liu Jicong 已提交
324
#if 0
L
Liu Jicong 已提交
325 326 327 328 329 330
      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 已提交
331 332 333
#endif
      tqReaderSetDataMsg(pReader, body, pReader->pWalReader->pHead->head.version);
#if 0
L
Liu Jicong 已提交
334
      }
L
Liu Jicong 已提交
335
#endif
L
Liu Jicong 已提交
336 337 338
    }

    while (tqNextDataBlock(pReader)) {
L
Liu Jicong 已提交
339
      // TODO mem free
L
Liu Jicong 已提交
340 341 342
      memset(&ret->data, 0, sizeof(SSDataBlock));
      int32_t code = tqRetrieveDataBlock(&ret->data, pReader);
      if (code != 0 || ret->data.info.rows == 0) {
L
Liu Jicong 已提交
343
        ASSERT(0);
L
Liu Jicong 已提交
344
        continue;
L
Liu Jicong 已提交
345 346
      }
      ret->fetchType = FETCH_TYPE__DATA;
L
Liu Jicong 已提交
347
      tqDebug("return data rows %d", ret->data.info.rows);
L
Liu Jicong 已提交
348 349 350 351
      return 0;
    }

    if (fromProcessedMsg) {
L
Liu Jicong 已提交
352 353
      ret->offset.type = TMQ_OFFSET__LOG;
      ret->offset.version = pReader->ver;
L
Liu Jicong 已提交
354
      ASSERT(pReader->ver >= 0);
L
Liu Jicong 已提交
355
      ret->fetchType = FETCH_TYPE__SEP;
S
Shengliang Guan 已提交
356
      tqDebug("return offset %" PRId64 ", processed finish", ret->offset.version);
L
Liu Jicong 已提交
357 358 359
      return 0;
    }
  }
L
Liu Jicong 已提交
360 361
}

L
Liu Jicong 已提交
362
int32_t tqReaderSetDataMsg(STqReader* pReader, const SSubmitReq* pMsg, int64_t ver) {
L
Liu Jicong 已提交
363
  pReader->pMsg = pMsg;
L
Liu Jicong 已提交
364

L
Liu Jicong 已提交
365
  if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1;
L
Liu Jicong 已提交
366
  while (true) {
L
Liu Jicong 已提交
367 368
    if (tGetSubmitMsgNext(&pReader->msgIter, &pReader->pBlock) < 0) return -1;
    if (pReader->pBlock == NULL) break;
L
Liu Jicong 已提交
369 370
  }

L
Liu Jicong 已提交
371 372 373
  if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1;
  pReader->ver = ver;
  memset(&pReader->blkIter, 0, sizeof(SSubmitBlkIter));
L
Liu Jicong 已提交
374
  return 0;
L
Liu Jicong 已提交
375 376
}

L
Liu Jicong 已提交
377 378
bool tqNextDataBlock(STqReader* pReader) {
  if (pReader->pMsg == NULL) return false;
L
Liu Jicong 已提交
379
  while (1) {
L
Liu Jicong 已提交
380
    if (tGetSubmitMsgNext(&pReader->msgIter, &pReader->pBlock) < 0) {
L
Liu Jicong 已提交
381 382
      return false;
    }
L
Liu Jicong 已提交
383 384
    if (pReader->pBlock == NULL) {
      pReader->pMsg = NULL;
5
54liuyao 已提交
385 386
      return false;
    }
L
Liu Jicong 已提交
387

L
Liu Jicong 已提交
388
    if (pReader->tbIdHash == NULL) {
L
Liu Jicong 已提交
389 390
      return true;
    }
L
Liu Jicong 已提交
391
    void* ret = taosHashGet(pReader->tbIdHash, &pReader->msgIter.uid, sizeof(int64_t));
S
Shengliang Guan 已提交
392
    /*tqDebug("search uid %" PRId64, pHandle->msgIter.uid);*/
L
Liu Jicong 已提交
393
    if (ret != NULL) {
S
Shengliang Guan 已提交
394
      /*tqDebug("find   uid %" PRId64, pHandle->msgIter.uid);*/
L
Liu Jicong 已提交
395
      return true;
396 397 398 399 400
    }
  }
  return false;
}

L
Liu Jicong 已提交
401
bool tqNextDataBlockFilterOut(STqReader* pHandle, SHashObj* filterOutUids) {
402 403 404 405 406 407 408 409 410 411
  while (1) {
    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
      return false;
    }
    if (pHandle->pBlock == NULL) return false;

    ASSERT(pHandle->tbIdHash == NULL);
    void* ret = taosHashGet(filterOutUids, &pHandle->msgIter.uid, sizeof(int64_t));
    if (ret == NULL) {
      return true;
L
Liu Jicong 已提交
412 413 414 415 416
    }
  }
  return false;
}

L
Liu Jicong 已提交
417
int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader) {
418
  // TODO: cache multiple schema
L
Liu Jicong 已提交
419 420 421 422
  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);
423
    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
L
Liu Jicong 已提交
424
    if (pReader->pSchema == NULL) {
S
Shengliang Guan 已提交
425
      tqWarn("cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64 "), version %d, possibly dropped table",
L
Liu Jicong 已提交
426
             pReader->msgIter.uid, pReader->msgIter.suid, pReader->cachedSchemaVer);
L
Liu Jicong 已提交
427
      /*ASSERT(0);*/
428
      pReader->cachedSchemaSuid = 0;
429
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
L
Liu Jicong 已提交
430 431
      return -1;
    }
L
Liu Jicong 已提交
432

L
Liu Jicong 已提交
433
    if (pReader->pSchemaWrapper) tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
434
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
L
Liu Jicong 已提交
435
    if (pReader->pSchemaWrapper == NULL) {
S
Shengliang Guan 已提交
436
      tqWarn("cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
L
Liu Jicong 已提交
437
             pReader->msgIter.uid, pReader->cachedSchemaVer);
L
Liu Jicong 已提交
438
      /*ASSERT(0);*/
439
      pReader->cachedSchemaSuid = 0;
L
Liu Jicong 已提交
440 441 442
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }
L
Liu Jicong 已提交
443 444
    pReader->cachedSchemaVer = sversion;
    pReader->cachedSchemaSuid = pReader->msgIter.suid;
L
Liu Jicong 已提交
445 446
  }

L
Liu Jicong 已提交
447 448
  STSchema*       pTschema = pReader->pSchema;
  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
L
Liu Jicong 已提交
449

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

L
Liu Jicong 已提交
452 453 454 455
  if (colNumNeed == 0) {
    int32_t colMeta = 0;
    while (colMeta < pSchemaWrapper->nCols) {
      SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
456
      SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
L
Liu Jicong 已提交
457
      int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
458
      if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
459
        goto FAIL;
L
Liu Jicong 已提交
460
      }
L
Liu Jicong 已提交
461
      colMeta++;
L
Liu Jicong 已提交
462 463 464 465 466 467 468 469 470 471 472
    }
  } 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 已提交
473
      col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, colNeed);
L
Liu Jicong 已提交
474 475 476 477 478
      if (colIdSchema < colIdNeed) {
        colMeta++;
      } else if (colIdSchema > colIdNeed) {
        colNeed++;
      } else {
479
        SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
L
Liu Jicong 已提交
480
        int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
481
        if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
482 483 484 485 486
          goto FAIL;
        }
        colMeta++;
        colNeed++;
      }
L
Liu Jicong 已提交
487 488
    }
  }
L
Liu Jicong 已提交
489

L
Liu Jicong 已提交
490
  if (blockDataEnsureCapacity(pBlock, pReader->msgIter.numOfRows) < 0) {
491
    terrno = TSDB_CODE_OUT_OF_MEMORY;
492 493 494 495
    goto FAIL;
  }

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

L
Liu Jicong 已提交
497 498 499 500
  STSRowIter iter = {0};
  tdSTSRowIterInit(&iter, pTschema);
  STSRow* row;
  int32_t curRow = 0;
501

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

L
Liu Jicong 已提交
504 505
  pBlock->info.uid = pReader->msgIter.uid;
  pBlock->info.rows = pReader->msgIter.numOfRows;
506
  pBlock->info.version = pReader->pMsg->version;
507

L
Liu Jicong 已提交
508
  while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) {
L
Liu Jicong 已提交
509 510
    tdSTSRowIterReset(&iter, row);
    // get all wanted col of that block
L
Liu Jicong 已提交
511
    for (int32_t i = 0; i < colActual; i++) {
512
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
L
Liu Jicong 已提交
513
      SCellVal         sVal = {0};
C
Cary Xu 已提交
514
      if (!tdSTSRowIterFetch(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
L
Liu Jicong 已提交
515 516
        break;
      }
517
      if (colDataAppend(pColData, curRow, sVal.val, sVal.valType != TD_VTYPE_NORM) < 0) {
L
Liu Jicong 已提交
518
        goto FAIL;
L
Liu Jicong 已提交
519
      }
L
Liu Jicong 已提交
520 521 522
    }
    curRow++;
  }
L
Liu Jicong 已提交
523
  return 0;
524

L
Liu Jicong 已提交
525
FAIL:
526
  blockDataFreeRes(pBlock);
L
Liu Jicong 已提交
527
  return -1;
L
Liu Jicong 已提交
528
}
H
Hongze Cheng 已提交
529

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

L
Liu Jicong 已提交
532 533 534 535
int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash) {
    taosHashClear(pReader->tbIdHash);
  } else {
5
54liuyao 已提交
536
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
H
Hongze Cheng 已提交
537 538
  }

L
Liu Jicong 已提交
539
  if (pReader->tbIdHash == NULL) {
H
Hongze Cheng 已提交
540 541 542 543 544 545
    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 已提交
546
    taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
H
Hongze Cheng 已提交
547 548 549 550 551
  }

  return 0;
}

L
Liu Jicong 已提交
552 553
int tqReaderAddTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash == NULL) {
5
54liuyao 已提交
554
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
555
    if (pReader->tbIdHash == NULL) {
H
Hongze Cheng 已提交
556 557 558 559 560 561 562
      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 已提交
563
    taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
H
Hongze Cheng 已提交
564 565 566 567
  }

  return 0;
}
568

L
Liu Jicong 已提交
569 570
int tqReaderRemoveTbUidList(STqReader* pReader, const SArray* tbUidList) {
  ASSERT(pReader->tbIdHash != NULL);
571

L
Liu Jicong 已提交
572 573
  for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
L
Liu Jicong 已提交
574
    taosHashRemove(pReader->tbIdHash, pKey, sizeof(int64_t));
575 576 577 578
  }

  return 0;
}
L
Liu Jicong 已提交
579 580 581 582

int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
  void* pIter = NULL;
  while (1) {
583
    pIter = taosHashIterate(pTq->pHandle, pIter);
L
Liu Jicong 已提交
584 585 586
    if (pIter == NULL) break;
    STqHandle* pExec = (STqHandle*)pIter;
    if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
587
      int32_t code = qUpdateQualifiedTableId(pExec->execHandle.task, tbUidList, isAdd);
L
Liu Jicong 已提交
588
      ASSERT(code == 0);
L
Liu Jicong 已提交
589 590 591 592 593
    } 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 已提交
594
          taosHashPut(pExec->execHandle.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0);
L
Liu Jicong 已提交
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
    } 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);

          int32_t code = metaGetTableEntryByUid(&mr, *id);
          if (code != TSDB_CODE_SUCCESS) {
            qError("failed to get table meta, uid:%" PRIu64 " code:%s", *id, tstrerror(terrno));
            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 已提交
628
    } else {
629
      ASSERT(0);
L
Liu Jicong 已提交
630 631 632
    }
  }
  while (1) {
L
Liu Jicong 已提交
633
    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
L
Liu Jicong 已提交
634
    if (pIter == NULL) break;
635
    SStreamTask* pTask = *(SStreamTask**)pIter;
636
    if (pTask->taskLevel == TASK_LEVEL__SOURCE) {
L
Liu Jicong 已提交
637 638 639 640 641 642
      int32_t code = qUpdateQualifiedTableId(pTask->exec.executor, tbUidList, isAdd);
      ASSERT(code == 0);
    }
  }
  return 0;
}