tqRead.c 25.2 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;
}

401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429
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 已提交
430
bool tqNextDataBlockFilterOut(STqReader* pHandle, SHashObj* filterOutUids) {
431 432 433 434 435 436 437 438 439 440
  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 已提交
441 442 443 444 445
    }
  }
  return false;
}

L
Liu Jicong 已提交
446
int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader) {
447
  // TODO: cache multiple schema
L
Liu Jicong 已提交
448 449 450 451
  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);
452
    pReader->pSchema = metaGetTbTSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
L
Liu Jicong 已提交
453
    if (pReader->pSchema == NULL) {
S
Shengliang Guan 已提交
454
      tqWarn("cannot found tsschema for table: uid:%" PRId64 " (suid:%" PRId64 "), version %d, possibly dropped table",
L
Liu Jicong 已提交
455
             pReader->msgIter.uid, pReader->msgIter.suid, pReader->cachedSchemaVer);
L
Liu Jicong 已提交
456
      /*ASSERT(0);*/
457
      pReader->cachedSchemaSuid = 0;
458
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
L
Liu Jicong 已提交
459 460
      return -1;
    }
L
Liu Jicong 已提交
461

L
Liu Jicong 已提交
462
    if (pReader->pSchemaWrapper) tDeleteSSchemaWrapper(pReader->pSchemaWrapper);
463
    pReader->pSchemaWrapper = metaGetTableSchema(pReader->pVnodeMeta, pReader->msgIter.uid, sversion, 1);
L
Liu Jicong 已提交
464
    if (pReader->pSchemaWrapper == NULL) {
S
Shengliang Guan 已提交
465
      tqWarn("cannot found schema wrapper for table: suid:%" PRId64 ", version %d, possibly dropped table",
L
Liu Jicong 已提交
466
             pReader->msgIter.uid, pReader->cachedSchemaVer);
L
Liu Jicong 已提交
467
      /*ASSERT(0);*/
468
      pReader->cachedSchemaSuid = 0;
L
Liu Jicong 已提交
469 470 471
      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
      return -1;
    }
L
Liu Jicong 已提交
472 473
    pReader->cachedSchemaVer = sversion;
    pReader->cachedSchemaSuid = pReader->msgIter.suid;
L
Liu Jicong 已提交
474 475
  }

L
Liu Jicong 已提交
476 477
  STSchema*       pTschema = pReader->pSchema;
  SSchemaWrapper* pSchemaWrapper = pReader->pSchemaWrapper;
L
Liu Jicong 已提交
478

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

L
Liu Jicong 已提交
481 482 483 484
  if (colNumNeed == 0) {
    int32_t colMeta = 0;
    while (colMeta < pSchemaWrapper->nCols) {
      SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
485
      SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
L
Liu Jicong 已提交
486
      int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
487
      if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
488
        goto FAIL;
L
Liu Jicong 已提交
489
      }
L
Liu Jicong 已提交
490
      colMeta++;
L
Liu Jicong 已提交
491 492 493 494 495 496 497 498 499 500 501
    }
  } 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 已提交
502
      col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pReader->pColIdList, colNeed);
L
Liu Jicong 已提交
503 504 505 506 507
      if (colIdSchema < colIdNeed) {
        colMeta++;
      } else if (colIdSchema > colIdNeed) {
        colNeed++;
      } else {
508
        SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
L
Liu Jicong 已提交
509
        int32_t         code = blockDataAppendColInfo(pBlock, &colInfo);
510
        if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
511 512 513 514 515
          goto FAIL;
        }
        colMeta++;
        colNeed++;
      }
L
Liu Jicong 已提交
516 517
    }
  }
L
Liu Jicong 已提交
518

L
Liu Jicong 已提交
519
  if (blockDataEnsureCapacity(pBlock, pReader->msgIter.numOfRows) < 0) {
520
    terrno = TSDB_CODE_OUT_OF_MEMORY;
521 522 523 524
    goto FAIL;
  }

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

L
Liu Jicong 已提交
526 527 528 529
  STSRowIter iter = {0};
  tdSTSRowIterInit(&iter, pTschema);
  STSRow* row;
  int32_t curRow = 0;
530

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

H
Haojun Liao 已提交
533
  pBlock->info.id.uid = pReader->msgIter.uid;
L
Liu Jicong 已提交
534
  pBlock->info.rows = pReader->msgIter.numOfRows;
535
  pBlock->info.version = pReader->pMsg->version;
536
  pBlock->info.dataLoad = 1;
537

L
Liu Jicong 已提交
538
  while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) {
L
Liu Jicong 已提交
539 540
    tdSTSRowIterReset(&iter, row);
    // get all wanted col of that block
L
Liu Jicong 已提交
541
    for (int32_t i = 0; i < colActual; i++) {
542
      SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
L
Liu Jicong 已提交
543
      SCellVal         sVal = {0};
C
Cary Xu 已提交
544
      if (!tdSTSRowIterFetch(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
L
Liu Jicong 已提交
545 546
        break;
      }
547
      if (colDataSetVal(pColData, curRow, sVal.val, sVal.valType != TD_VTYPE_NORM) < 0) {
L
Liu Jicong 已提交
548
        goto FAIL;
L
Liu Jicong 已提交
549
      }
L
Liu Jicong 已提交
550 551 552
    }
    curRow++;
  }
L
Liu Jicong 已提交
553
  return 0;
554

L
Liu Jicong 已提交
555
FAIL:
556
  blockDataFreeRes(pBlock);
L
Liu Jicong 已提交
557
  return -1;
L
Liu Jicong 已提交
558
}
H
Hongze Cheng 已提交
559

560
int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas) {
561 562 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
  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);
      /*ASSERT(0);*/
      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);
      /*ASSERT(0);*/
      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;
596
  int32_t lastRow = 0;
597 598

  char* assigned = taosMemoryCalloc(1, pSchemaWrapper->nCols);
599
  if (assigned == NULL) return -1;
600 601 602 603 604 605 606 607 608 609

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

610
    tqDebug("vgId:%d, row of block %d", pReader->pWalReader->pWal->cfg.vgId, curRow);
611 612 613 614 615
    for (int32_t i = 0; i < colAtMost; i++) {
      SCellVal sVal = {0};
      if (!tdSTSRowIterFetch(&iter, pSchemaWrapper->pSchema[i].colId, pSchemaWrapper->pSchema[i].type, &sVal)) {
        break;
      }
616
      tqDebug("vgId:%d, %d col, type %d", pReader->pWalReader->pWal->cfg.vgId, i, sVal.valType);
617 618 619 620 621 622 623 624 625 626 627 628 629
      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) {
630 631 632 633 634 635 636 637 638
      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);
639 640
        goto FAIL;
      }
641 642 643 644 645 646
      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));
647 648

      taosArrayPush(blocks, &block);
649
      taosArrayPush(schemas, &pSW);
650 651 652
    }

    SSDataBlock* pBlock = taosArrayGetLast(blocks);
H
Haojun Liao 已提交
653
    pBlock->info.id.uid = pReader->msgIter.uid;
654
    pBlock->info.rows = 0;
655 656
    pBlock->info.version = pReader->pMsg->version;

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

660 661 662 663 664
    if (blockDataEnsureCapacity(pBlock, pReader->msgIter.numOfRows - curRow) < 0) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto FAIL;
    }

665
    tdSTSRowIterReset(&iter, row);
666 667 668 669 670 671 672 673 674 675
    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;
      }

      ASSERT(sVal.valType != TD_VTYPE_NONE);

676
      if (colDataSetVal(pColData, curRow, sVal.val, sVal.valType == TD_VTYPE_NULL) < 0) {
677 678
        goto FAIL;
      }
679 680
      tqDebug("vgId:%d, row %d col %d append %d", pReader->pWalReader->pWal->cfg.vgId, curRow, i,
              sVal.valType == TD_VTYPE_NULL);
681 682 683
    }
    curRow++;
  }
684 685
  SSDataBlock* pLastBlock = taosArrayGetLast(blocks);
  pLastBlock->info.rows = curRow - lastRow;
686 687 688 689 690 691 692 693 694

  taosMemoryFree(assigned);
  return 0;

FAIL:
  taosMemoryFree(assigned);
  return -1;
}

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

L
Liu Jicong 已提交
697 698 699 700
int tqReaderSetTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash) {
    taosHashClear(pReader->tbIdHash);
  } else {
5
54liuyao 已提交
701
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
H
Hongze Cheng 已提交
702 703
  }

L
Liu Jicong 已提交
704
  if (pReader->tbIdHash == NULL) {
H
Hongze Cheng 已提交
705 706 707 708 709 710
    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 已提交
711
    taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
H
Hongze Cheng 已提交
712 713 714 715 716
  }

  return 0;
}

L
Liu Jicong 已提交
717 718
int tqReaderAddTbUidList(STqReader* pReader, const SArray* tbUidList) {
  if (pReader->tbIdHash == NULL) {
5
54liuyao 已提交
719
    pReader->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
L
Liu Jicong 已提交
720
    if (pReader->tbIdHash == NULL) {
H
Hongze Cheng 已提交
721 722 723 724 725 726 727
      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 已提交
728
    taosHashPut(pReader->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
H
Hongze Cheng 已提交
729 730 731 732
  }

  return 0;
}
733

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

L
Liu Jicong 已提交
737 738
  for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
    int64_t* pKey = (int64_t*)taosArrayGet(tbUidList, i);
L
Liu Jicong 已提交
739
    taosHashRemove(pReader->tbIdHash, pKey, sizeof(int64_t));
740 741 742 743
  }

  return 0;
}
L
Liu Jicong 已提交
744 745 746 747

int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
  void* pIter = NULL;
  while (1) {
748
    pIter = taosHashIterate(pTq->pHandle, pIter);
H
Haojun Liao 已提交
749 750 751 752
    if (pIter == NULL) {
      break;
    }

L
Liu Jicong 已提交
753 754
    STqHandle* pExec = (STqHandle*)pIter;
    if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
755
      int32_t code = qUpdateQualifiedTableId(pExec->execHandle.task, tbUidList, isAdd);
L
Liu Jicong 已提交
756
      ASSERT(code == 0);
L
Liu Jicong 已提交
757 758 759 760 761
    } 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 已提交
762
          taosHashPut(pExec->execHandle.execDb.pFilterOutTbUid, &tbUid, sizeof(int64_t), NULL, 0);
L
Liu Jicong 已提交
763 764
        }
      }
765 766 767 768 769 770 771 772
    } 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 已提交
773
          int32_t code = metaGetTableEntryByUidCache(&mr, *id);
774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795
          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 已提交
796
    } else {
797
      ASSERT(0);
L
Liu Jicong 已提交
798 799 800
    }
  }
  while (1) {
L
Liu Jicong 已提交
801
    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
L
Liu Jicong 已提交
802
    if (pIter == NULL) break;
803
    SStreamTask* pTask = *(SStreamTask**)pIter;
804
    if (pTask->taskLevel == TASK_LEVEL__SOURCE) {
L
Liu Jicong 已提交
805 806 807 808 809 810
      int32_t code = qUpdateQualifiedTableId(pTask->exec.executor, tbUidList, isAdd);
      ASSERT(code == 0);
    }
  }
  return 0;
}