taosdata/TDengine | Build 3798 | source/libs/executor/src/executil.c | Coveralls

59

static int64_t getLimit(const SNode* pLimit) { return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->limit) ? -1 : ((SLimitNode*)pLimit)->limit->datum.i; }

×

60

static int64_t getOffset(const SNode* pLimit) { return (NULL == pLimit || NULL == ((SLimitNode*)pLimit)->offset) ? -1 : ((SLimitNode*)pLimit)->offset->datum.i; }

×

63

void initResultRowInfo(SResultRowInfo* pResultRowInfo) {

×

64

  pResultRowInfo->size = 0;

×

65

  pResultRowInfo->cur.pageId = -1;

×

66

×

68

void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; }

×

70

void resetResultRow(SResultRow* pResultRow, size_t entrySize) {

×

71

  pResultRow->numOfRows = 0;

×

72

  pResultRow->closed = false;

×

73

  pResultRow->endInterp = false;

×

74

  pResultRow->startInterp = false;

×

76

  if (entrySize > 0) {

×

77

    memset(pResultRow->pEntryInfo, 0, entrySize);

×

79

×

82

SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) {

×

83

  return (SResultRowEntryInfo*)((char*)pRow->pEntryInfo + offset[index]);

×

86

size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {

×

87

  int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow);

×

89

  for (int32_t i = 0; i < numOfOutput; ++i) {

×

90

    rowSize += pCtx[i].resDataInfo.interBufSize;

×

93

  return rowSize;

×

97

int32_t getResultRowFromBuf(SExprSupp *pSup, const char* inBuf, size_t inBufSize, char **outBuf, size_t *outBufSize) {

×

98

  if (inBuf == NULL || pSup == NULL) {

×

99

    qError("invalid input parameters, inBuf:%p, pSup:%p", inBuf, pSup);

×

100

    return TSDB_CODE_INVALID_PARA;

×

102

  SqlFunctionCtx *pCtx = pSup->pCtx;

×

103

  int32_t        *offset = pSup->rowEntryInfoOffset;

×

104

  SResultRow     *pResultRow  = NULL;

×

105

  size_t          processedSize = 0;

×

106

  int32_t         code = TSDB_CODE_SUCCESS;

×

109

  *outBufSize = getResultRowSize(pCtx, pSup->numOfExprs);

×

110

  *outBuf = taosMemoryMalloc(*outBufSize);

×

111

  if (*outBuf == NULL) {

×

112

    qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);

×

113

    return terrno;

×

115

  pResultRow = (SResultRow*)*outBuf;

×

116

  (void)memcpy(pResultRow, inBuf, sizeof(SResultRow));

×

117

  inBuf += sizeof(SResultRow);

×

118

  processedSize += sizeof(SResultRow);

×

120

  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {

×

121

    int32_t len = *(int32_t*)inBuf;

×

122

    inBuf += sizeof(int32_t);

×

123

    processedSize += sizeof(int32_t);

×

124

    if (pResultRow->version != FUNCTION_RESULT_INFO_VERSION && pCtx->fpSet.decode) {

×

125

      code = pCtx->fpSet.decode(&pCtx[i], inBuf, getResultEntryInfo(pResultRow, i, offset), pResultRow->version);

×

126

      if (code != TSDB_CODE_SUCCESS) {

×

127

        qError("failed to decode result row, code:%d", code);

×

128

        return code;

×

131

      (void)memcpy(getResultEntryInfo(pResultRow, i, offset), inBuf, len);

×

133

    inBuf += len;

×

134

    processedSize += len;

×

137

  if (processedSize < inBufSize) {

×

139

    size_t leftLen = inBufSize - processedSize;

×

140

    TAOS_MEMORY_REALLOC(*outBuf, *outBufSize + leftLen);

×

141

    if (*outBuf == NULL) {

×

142

      qError("failed to reallocate memory for output buffer, size:%zu", *outBufSize + leftLen);

×

143

      return terrno;

×

145

    (void)memcpy(*outBuf + *outBufSize, inBuf, leftLen);

×

146

    inBuf += leftLen;

×

147

    processedSize += leftLen;

×

148

    *outBufSize += leftLen;

×

150

  return TSDB_CODE_SUCCESS;

×

154

int32_t putResultRowToBuf(SExprSupp *pSup, const char* inBuf, size_t inBufSize, char **outBuf, size_t *outBufSize) {

×

155

  if (pSup == NULL || inBuf == NULL || outBuf == NULL || outBufSize == NULL) {

×

156

    qError("invalid input parameters, inBuf:%p, pSup:%p, outBufSize:%p, outBuf:%p", inBuf, pSup, outBufSize, outBuf);

×

157

    return TSDB_CODE_INVALID_PARA;

×

160

  SqlFunctionCtx *pCtx = pSup->pCtx;

×

161

  int32_t        *offset = pSup->rowEntryInfoOffset;

×

162

  SResultRow     *pResultRow = (SResultRow*)inBuf;

×

163

  size_t          rowSize = getResultRowSize(pCtx, pSup->numOfExprs);

×

165

  if (rowSize > inBufSize) {

×

166

    qError("invalid input buffer size, rowSize:%zu, inBufSize:%zu", rowSize, inBufSize);

×

167

    return TSDB_CODE_INVALID_PARA;

×

171

  *outBufSize = rowSize + sizeof(int32_t) * pSup->numOfExprs;

×

172

  if (rowSize < inBufSize) {

×

173

    *outBufSize += inBufSize - rowSize;

×

176

  *outBuf = taosMemoryMalloc(*outBufSize);

×

177

  if (*outBuf == NULL) {

×

178

    qError("failed to allocate memory for output buffer, size:%zu", *outBufSize);

×

179

    return terrno;

×

182

  char *pBuf = *outBuf;

×

183

  pResultRow->version = FUNCTION_RESULT_INFO_VERSION;

×

184

  (void)memcpy(pBuf, pResultRow, sizeof(SResultRow));

×

185

  pBuf += sizeof(SResultRow);

×

186

  for (int32_t i = 0; i < pSup->numOfExprs; ++i) {

×

187

    size_t len = sizeof(SResultRowEntryInfo) + pCtx[i].resDataInfo.interBufSize;

×

188

    *(int32_t *) pBuf = (int32_t)len;

×

189

    pBuf += sizeof(int32_t);

×

190

    (void)memcpy(pBuf, getResultEntryInfo(pResultRow, i, offset), len);

×

191

    pBuf += len;

×

194

  if (rowSize < inBufSize) {

×

196

    size_t leftLen = inBufSize - rowSize;

×

197

    (void)memcpy(pBuf, inBuf + rowSize, leftLen);

×

198

    pBuf += leftLen;

×

200

  return TSDB_CODE_SUCCESS;

×

203

static void freeEx(void* p) { taosMemoryFree(*(void**)p); }

×

205

void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {

×

206

  taosMemoryFreeClear(pGroupResInfo->pBuf);

×

207

  if (pGroupResInfo->freeItem) {

×

209

    taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);

×

210

    pGroupResInfo->freeItem = false;

×

211

    pGroupResInfo->pRows = NULL;

×

213

    taosArrayDestroy(pGroupResInfo->pRows);

×

214

    pGroupResInfo->pRows = NULL;

×

216

  pGroupResInfo->index = 0;

×

217

  pGroupResInfo->delIndex = 0;

×

218

×

220

int32_t resultrowComparAsc(const void* p1, const void* p2) {

×

221

  SResKeyPos* pp1 = *(SResKeyPos**)p1;

×

222

  SResKeyPos* pp2 = *(SResKeyPos**)p2;

×

224

  if (pp1->groupId == pp2->groupId) {

×

225

    int64_t pts1 = *(int64_t*)pp1->key;

×

226

    int64_t pts2 = *(int64_t*)pp2->key;

×

228

    if (pts1 == pts2) {

×

229

      return 0;

×

231

      return pts1 < pts2 ? -1 : 1;

×

234

    return pp1->groupId < pp2->groupId ? -1 : 1;

×

238

static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); }

×

240

int32_t initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SSHashObj* pHashmap, int32_t order) {

×

241

  int32_t code = TSDB_CODE_SUCCESS;

×

242

  int32_t lino = 0;

×

243

  if (pGroupResInfo->pRows != NULL) {

×

244

    taosArrayDestroy(pGroupResInfo->pRows);

×

246

  if (pGroupResInfo->pBuf) {

×

247

    taosMemoryFree(pGroupResInfo->pBuf);

×

248

    pGroupResInfo->pBuf = NULL;

×

252

  int32_t size = tSimpleHashGetSize(pHashmap);

×

254

  void* pData = NULL;

×

255

  pGroupResInfo->pRows = taosArrayInit(size, POINTER_BYTES);

×

256

  QUERY_CHECK_NULL(pGroupResInfo->pRows, code, lino, _end, terrno);

×

258

  size_t  keyLen = 0;

×

259

  int32_t iter = 0;

×

260

  int64_t bufLen = 0, offset = 0;

×

263

  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {

×

265

    bufLen += keyLen + sizeof(SResultRowPosition);

×

268

  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);

×

269

  QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);

×

271

  iter = 0;

×

272

  while ((pData = tSimpleHashIterate(pHashmap, pData, &iter)) != NULL) {

×

273

    void* key = tSimpleHashGetKey(pData, &keyLen);

×

275

    SResKeyPos* p = (SResKeyPos*)(pGroupResInfo->pBuf + offset);

×

277

    p->groupId = *(uint64_t*)key;

×

278

    p->pos = *(SResultRowPosition*)pData;

×

279

    memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t));

×

280

    void* tmp = taosArrayPush(pGroupResInfo->pRows, &p);

×

281

    QUERY_CHECK_NULL(pGroupResInfo->pBuf, code, lino, _end, terrno);

×

283

    offset += keyLen + sizeof(struct SResultRowPosition);

×

286

  if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) {

×

287

    __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc;

×

288

    size = POINTER_BYTES;

×

289

    taosSort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), size, fn);

×

292

  pGroupResInfo->index = 0;

×

294

_end:

×

295

  if (code != TSDB_CODE_SUCCESS) {

×

296

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

298

  return code;

×

301

void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList) {

×

302

  if (pGroupResInfo->pRows != NULL) {

×

303

    taosArrayDestroy(pGroupResInfo->pRows);

×

306

  pGroupResInfo->freeItem = true;

×

307

  pGroupResInfo->pRows = pArrayList;

×

308

  pGroupResInfo->index = 0;

×

309

  pGroupResInfo->delIndex = 0;

×

310

×

312

bool hasRemainResults(SGroupResInfo* pGroupResInfo) {

×

313

  if (pGroupResInfo->pRows == NULL) {

×

314

    return false;

×

317

  return pGroupResInfo->index < taosArrayGetSize(pGroupResInfo->pRows);

×

320

int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {

×

321

  if (pGroupResInfo->pRows == 0) {

×

322

    return 0;

×

325

  return (int32_t)taosArrayGetSize(pGroupResInfo->pRows);

×

328

SArray* createSortInfo(SNodeList* pNodeList) {

×

329

  size_t numOfCols = 0;

×

331

  if (pNodeList != NULL) {

×

332

    numOfCols = LIST_LENGTH(pNodeList);

×

334

    numOfCols = 0;

×

337

  SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));

×

338

  if (pList == NULL) {

×

339

    return pList;

×

342

  for (int32_t i = 0; i < numOfCols; ++i) {

×

343

    SOrderByExprNode* pSortKey = (SOrderByExprNode*)nodesListGetNode(pNodeList, i);

×

344

    if (!pSortKey) {

×

345

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

346

      taosArrayDestroy(pList);

×

347

      pList = NULL;

×

348

      terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

349

      break;

×

351

    SBlockOrderInfo   bi = {0};

×

352

    bi.order = (pSortKey->order == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;

×

353

    bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST);

×

355

    SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr;

×

356

    bi.slotId = pColNode->slotId;

×

357

    void* tmp = taosArrayPush(pList, &bi);

×

358

    if (!tmp) {

×

359

      taosArrayDestroy(pList);

×

360

      pList = NULL;

×

361

      break;

×

365

  return pList;

×

368

SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {

×

369

  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);

×

370

  SSDataBlock* pBlock = NULL;

×

371

  int32_t      code = createDataBlock(&pBlock);

×

372

  if (code) {

×

373

    terrno = code;

×

374

    return NULL;

×

377

  pBlock->info.id.blockId = pNode->dataBlockId;

×

378

  pBlock->info.type = STREAM_INVALID;

×

379

  pBlock->info.calWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};

×

380

  pBlock->info.watermark = INT64_MIN;

×

382

  for (int32_t i = 0; i < numOfCols; ++i) {

×

383

    SSlotDescNode*  pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);

×

384

    if (!pDescNode) {

×

385

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

386

      blockDataDestroy(pBlock);

×

387

      pBlock = NULL;

×

388

      terrno = TSDB_CODE_INVALID_PARA;

×

389

      break;

×

392

        createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);

×

393

    idata.info.scale = pDescNode->dataType.scale;

×

394

    idata.info.precision = pDescNode->dataType.precision;

×

395

    idata.info.noData = pDescNode->reserve;

×

397

    code = blockDataAppendColInfo(pBlock, &idata);

×

398

    if (code != TSDB_CODE_SUCCESS) {

×

399

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

400

      blockDataDestroy(pBlock);

×

401

      pBlock = NULL;

×

402

      terrno = code;

×

403

      break;

×

407

  return pBlock;

×

410

int32_t prepareDataBlockBuf(SSDataBlock* pDataBlock, SColMatchInfo* pMatchInfo) {

×

411

  SDataBlockInfo* pBlockInfo = &pDataBlock->info;

×

413

  for (int32_t i = 0; i < taosArrayGetSize(pMatchInfo->pList); ++i) {

×

414

    SColMatchItem* pItem = taosArrayGet(pMatchInfo->pList, i);

×

415

    if (!pItem) {

×

416

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

417

      return terrno;

×

420

    if (pItem->isPk) {

×

421

      SColumnInfoData* pInfoData = taosArrayGet(pDataBlock->pDataBlock, pItem->dstSlotId);

×

422

      if (!pInfoData) {

×

423

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

424

        return terrno;

×

426

      pBlockInfo->pks[0].type = pInfoData->info.type;

×

427

      pBlockInfo->pks[1].type = pInfoData->info.type;

×

430

      if (IS_VAR_DATA_TYPE(pItem->dataType.type)) {

×

431

        pBlockInfo->pks[0].pData = taosMemoryCalloc(1, pInfoData->info.bytes);

×

432

        if (pBlockInfo->pks[0].pData == NULL) {

×

433

          return terrno;

×

436

        pBlockInfo->pks[1].pData = taosMemoryCalloc(1, pInfoData->info.bytes);

×

437

        if (pBlockInfo->pks[1].pData == NULL) {

×

438

          taosMemoryFreeClear(pBlockInfo->pks[0].pData);

×

439

          return terrno;

×

442

        pBlockInfo->pks[0].nData = pInfoData->info.bytes;

×

443

        pBlockInfo->pks[1].nData = pInfoData->info.bytes;

×

446

      break;

×

450

  return TSDB_CODE_SUCCESS;

×

453

EDealRes doTranslateTagExpr(SNode** pNode, void* pContext) {

×

454

  STransTagExprCtx* pCtx = pContext;

×

455

  SMetaReader*      mr = pCtx->pReader;

×

456

  bool              isTagCol = false, isTbname = false;

×

457

  if (nodeType(*pNode) == QUERY_NODE_COLUMN) {

×

458

    SColumnNode* pCol = (SColumnNode*)*pNode;

×

459

    if (pCol->colType == COLUMN_TYPE_TBNAME)

×

460

      isTbname = true;

×

462

      isTagCol = true;

×

463

  } else if (nodeType(*pNode) == QUERY_NODE_FUNCTION) {

×

464

    SFunctionNode* pFunc = (SFunctionNode*)*pNode;

×

465

    if (pFunc->funcType == FUNCTION_TYPE_TBNAME) isTbname = true;

×

467

  if (isTagCol) {

×

468

    SColumnNode* pSColumnNode = *(SColumnNode**)pNode;

×

470

    SValueNode* res = NULL;

×

471

    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);

×

472

    if (NULL == res) {

×

473

      return DEAL_RES_ERROR;

×

476

    res->translate = true;

×

477

    res->node.resType = pSColumnNode->node.resType;

×

479

    STagVal tagVal = {0};

×

480

    tagVal.cid = pSColumnNode->colId;

×

481

    const char* p = mr->pAPI->extractTagVal(mr->me.ctbEntry.pTags, pSColumnNode->node.resType.type, &tagVal);

×

482

    if (p == NULL) {

×

483

      res->node.resType.type = TSDB_DATA_TYPE_NULL;

×

484

    } else if (pSColumnNode->node.resType.type == TSDB_DATA_TYPE_JSON) {

×

485

      int32_t len = ((const STag*)p)->len;

×

486

      res->datum.p = taosMemoryCalloc(len + 1, 1);

×

487

      if (NULL == res->datum.p) {

×

488

        return DEAL_RES_ERROR;

×

490

      memcpy(res->datum.p, p, len);

×

491

    } else if (IS_VAR_DATA_TYPE(pSColumnNode->node.resType.type)) {

×

492

      res->datum.p = taosMemoryCalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1, 1);

×

493

      if (NULL == res->datum.p) {

×

494

        return DEAL_RES_ERROR;

×

496

      memcpy(varDataVal(res->datum.p), tagVal.pData, tagVal.nData);

×

497

      varDataSetLen(res->datum.p, tagVal.nData);

×

499

      int32_t code = nodesSetValueNodeValue(res, &(tagVal.i64));

×

500

      if (code != TSDB_CODE_SUCCESS) {

×

501

        return DEAL_RES_ERROR;

×

504

    nodesDestroyNode(*pNode);

×

505

    *pNode = (SNode*)res;

×

506

  } else if (isTbname) {

×

507

    SValueNode* res = NULL;

×

508

    pCtx->code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);

×

509

    if (NULL == res) {

×

510

      return DEAL_RES_ERROR;

×

513

    res->translate = true;

×

514

    res->node.resType = ((SExprNode*)(*pNode))->resType;

×

516

    int32_t len = strlen(mr->me.name);

×

517

    res->datum.p = taosMemoryCalloc(len + VARSTR_HEADER_SIZE + 1, 1);

×

518

    if (NULL == res->datum.p) {

×

519

      return DEAL_RES_ERROR;

×

521

    memcpy(varDataVal(res->datum.p), mr->me.name, len);

×

522

    varDataSetLen(res->datum.p, len);

×

523

    nodesDestroyNode(*pNode);

×

524

    *pNode = (SNode*)res;

×

527

  return DEAL_RES_CONTINUE;

×

530

int32_t isQualifiedTable(STableKeyInfo* info, SNode* pTagCond, void* metaHandle, bool* pQualified, SStorageAPI* pAPI) {

×

531

  int32_t     code = TSDB_CODE_SUCCESS;

×

532

  SMetaReader mr = {0};

×

534

  pAPI->metaReaderFn.initReader(&mr, metaHandle, META_READER_LOCK, &pAPI->metaFn);

×

535

  code = pAPI->metaReaderFn.getEntryGetUidCache(&mr, info->uid);

×

536

  if (TSDB_CODE_SUCCESS != code) {

×

537

    pAPI->metaReaderFn.clearReader(&mr);

×

538

    *pQualified = false;

×

540

    return TSDB_CODE_SUCCESS;

×

543

  SNode* pTagCondTmp = NULL;

×

544

  code = nodesCloneNode(pTagCond, &pTagCondTmp);

×

545

  if (TSDB_CODE_SUCCESS != code) {

×

546

    *pQualified = false;

×

547

    pAPI->metaReaderFn.clearReader(&mr);

×

548

    return code;

×

550

  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};

×

551

  nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);

×

552

  pAPI->metaReaderFn.clearReader(&mr);

×

553

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

554

    *pQualified = false;

×

555

    terrno = code;

×

556

    return code;

×

559

  SNode* pNew = NULL;

×

560

  code = scalarCalculateConstants(pTagCondTmp, &pNew);

×

561

  if (TSDB_CODE_SUCCESS != code) {

×

562

    terrno = code;

×

563

    nodesDestroyNode(pTagCondTmp);

×

564

    *pQualified = false;

×

566

    return code;

×

569

  SValueNode* pValue = (SValueNode*)pNew;

×

570

  *pQualified = pValue->datum.b;

×

572

  nodesDestroyNode(pNew);

×

573

  return TSDB_CODE_SUCCESS;

×

576

static EDealRes getColumn(SNode** pNode, void* pContext) {

×

577

  tagFilterAssist* pData = (tagFilterAssist*)pContext;

×

578

  SColumnNode*     pSColumnNode = NULL;

×

579

  if (QUERY_NODE_COLUMN == nodeType((*pNode))) {

×

580

    pSColumnNode = *(SColumnNode**)pNode;

×

581

  } else if (QUERY_NODE_FUNCTION == nodeType((*pNode))) {

×

582

    SFunctionNode* pFuncNode = *(SFunctionNode**)(pNode);

×

583

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

×

584

      pData->code = nodesMakeNode(QUERY_NODE_COLUMN, (SNode**)&pSColumnNode);

×

585

      if (NULL == pSColumnNode) {

×

586

        return DEAL_RES_ERROR;

×

588

      pSColumnNode->colId = -1;

×

589

      pSColumnNode->colType = COLUMN_TYPE_TBNAME;

×

590

      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;

×

591

      pSColumnNode->node.resType.bytes = TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE;

×

592

      nodesDestroyNode(*pNode);

×

593

      *pNode = (SNode*)pSColumnNode;

×

595

      return DEAL_RES_CONTINUE;

×

598

    return DEAL_RES_CONTINUE;

×

601

  void* data = taosHashGet(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId));

×

602

  if (!data) {

×

604

        taosHashPut(pData->colHash, &pSColumnNode->colId, sizeof(pSColumnNode->colId), pNode, sizeof((*pNode)));

×

605

    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {

×

606

      return DEAL_RES_ERROR;

×

608

    pSColumnNode->slotId = pData->index++;

×

609

    SColumnInfo cInfo = {.colId = pSColumnNode->colId,

×

610

                         .type = pSColumnNode->node.resType.type,

×

611

                         .bytes = pSColumnNode->node.resType.bytes,

×

612

                         .pk = pSColumnNode->isPk};

×

616

    void* tmp = taosArrayPush(pData->cInfoList, &cInfo);

×

617

    if (!tmp) {

×

618

      return DEAL_RES_ERROR;

×

621

    SColumnNode* col = *(SColumnNode**)data;

×

622

    pSColumnNode->slotId = col->slotId;

×

625

  return DEAL_RES_CONTINUE;

×

628

static int32_t createResultData(SDataType* pType, int32_t numOfRows, SScalarParam* pParam) {

×

629

  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));

×

630

  if (pColumnData == NULL) {

×

631

    return terrno;

×

634

  pColumnData->info.type = pType->type;

×

635

  pColumnData->info.bytes = pType->bytes;

×

636

  pColumnData->info.scale = pType->scale;

×

637

  pColumnData->info.precision = pType->precision;

×

639

  int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows, true);

×

640

  if (code != TSDB_CODE_SUCCESS) {

×

641

    terrno = code;

×

642

    releaseColInfoData(pColumnData);

×

643

    return terrno;

×

646

  pParam->columnData = pColumnData;

×

647

  pParam->colAlloced = true;

×

648

  return TSDB_CODE_SUCCESS;

×

651

static void releaseColInfoData(void* pCol) {

×

652

  if (pCol) {

×

653

    SColumnInfoData* col = (SColumnInfoData*)pCol;

×

654

    colDataDestroy(col);

×

655

    taosMemoryFree(col);

×

657

×

659

void freeItem(void* p) {

×

660

  STUidTagInfo* pInfo = p;

×

661

  if (pInfo->pTagVal != NULL) {

×

662

    taosMemoryFree(pInfo->pTagVal);

×

664

×

666

static int32_t genTagFilterDigest(const SNode* pTagCond, T_MD5_CTX* pContext) {

×

667

  if (pTagCond == NULL) {

×

668

    return TSDB_CODE_SUCCESS;

×

671

  char*   payload = NULL;

×

672

  int32_t len = 0;

×

673

  int32_t code = nodesNodeToMsg(pTagCond, &payload, &len);

×

674

  if (code != TSDB_CODE_SUCCESS) {

×

675

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

676

    return code;

×

679

  tMD5Init(pContext);

×

680

  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);

×

681

  tMD5Final(pContext);

×

683

  taosMemoryFree(payload);

×

684

  return TSDB_CODE_SUCCESS;

×

687

static int32_t genTbGroupDigest(const SNode* pGroup, uint8_t* filterDigest, T_MD5_CTX* pContext) {

×

688

  int32_t code = TSDB_CODE_SUCCESS;

×

689

  int32_t lino = 0;

×

690

  char*   payload = NULL;

×

691

  int32_t len = 0;

×

692

  code = nodesNodeToMsg(pGroup, &payload, &len);

×

693

  QUERY_CHECK_CODE(code, lino, _end);

×

695

  if (filterDigest[0]) {

×

696

    payload = taosMemoryRealloc(payload, len + tListLen(pContext->digest));

×

697

    QUERY_CHECK_NULL(payload, code, lino, _end, terrno);

×

698

    memcpy(payload + len, filterDigest + 1, tListLen(pContext->digest));

×

699

    len += tListLen(pContext->digest);

×

702

  tMD5Init(pContext);

×

703

  tMD5Update(pContext, (uint8_t*)payload, (uint32_t)len);

×

704

  tMD5Final(pContext);

×

706

_end:

×

707

  taosMemoryFree(payload);

×

708

  if (code != TSDB_CODE_SUCCESS) {

×

709

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

711

  return code;

×

714

int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInfo* pTableListInfo, uint8_t* digest,

×

716

  int32_t      code = TSDB_CODE_SUCCESS;

×

717

  int32_t      lino = 0;

×

718

  SArray*      pBlockList = NULL;

×

719

  SSDataBlock* pResBlock = NULL;

×

720

  void*        keyBuf = NULL;

×

721

  SArray*      groupData = NULL;

×

722

  SArray*      pUidTagList = NULL;

×

723

  SArray*      tableList = NULL;

×

725

  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);

×

726

  if (rows == 0) {

×

727

    return TSDB_CODE_SUCCESS;

×

730

  tagFilterAssist ctx = {0};

×

731

  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);

×

732

  if (ctx.colHash == NULL) {

×

733

    code = terrno;

×

734

    goto end;

×

737

  ctx.index = 0;

×

738

  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

×

739

  if (ctx.cInfoList == NULL) {

×

740

    code = terrno;

×

741

    goto end;

×

744

  SNode* pNode = NULL;

×

745

  FOREACH(pNode, group) {

×

746

    nodesRewriteExprPostOrder(&pNode, getColumn, (void*)&ctx);

×

747

    if (TSDB_CODE_SUCCESS != ctx.code) {

×

748

      code = ctx.code;

×

749

      goto end;

×

751

    REPLACE_NODE(pNode);

×

754

  T_MD5_CTX context = {0};

×

755

  if (tsTagFilterCache) {

×

756

    SNodeListNode* listNode = NULL;

×

757

    code = nodesMakeNode(QUERY_NODE_NODE_LIST, (SNode**)&listNode);

×

758

    if (TSDB_CODE_SUCCESS != code) {

×

759

      goto end;

×

761

    listNode->pNodeList = group;

×

762

    code = genTbGroupDigest((SNode*)listNode, digest, &context);

×

763

    QUERY_CHECK_CODE(code, lino, end);

×

765

    nodesFree(listNode);

×

767

    code = pAPI->metaFn.metaGetCachedTbGroup(pVnode, pTableListInfo->idInfo.suid, context.digest,

×

769

    QUERY_CHECK_CODE(code, lino, end);

×

771

    if (tableList) {

×

772

      taosArrayDestroy(pTableListInfo->pTableList);

×

773

      pTableListInfo->pTableList = tableList;

×

774

      qDebug("retrieve tb group list from cache, numOfTables:%d",

×

776

      goto end;

×

780

  pUidTagList = taosArrayInit(8, sizeof(STUidTagInfo));

×

781

  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);

×

783

  for (int32_t i = 0; i < rows; ++i) {

×

784

    STableKeyInfo* pkeyInfo = taosArrayGet(pTableListInfo->pTableList, i);

×

785

    QUERY_CHECK_NULL(pkeyInfo, code, lino, end, terrno);

×

786

    STUidTagInfo   info = {.uid = pkeyInfo->uid};

×

787

    void*          tmp = taosArrayPush(pUidTagList, &info);

×

788

    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);

×

791

  code = pAPI->metaFn.getTableTags(pVnode, pTableListInfo->idInfo.suid, pUidTagList);

×

792

  if (code != TSDB_CODE_SUCCESS) {

×

793

    goto end;

×

796

  int32_t numOfTables = taosArrayGetSize(pUidTagList);

×

797

  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);

×

798

  if (pResBlock == NULL) {

×

799

    code = terrno;

×

800

    goto end;

×

806

  pBlockList = taosArrayInit(2, POINTER_BYTES);

×

807

  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);

×

809

  void* tmp = taosArrayPush(pBlockList, &pResBlock);

×

810

  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);

×

812

  groupData = taosArrayInit(2, POINTER_BYTES);

×

813

  QUERY_CHECK_NULL(groupData, code, lino, end, terrno);

×

815

  FOREACH(pNode, group) {

×

816

    SScalarParam output = {0};

×

818

    switch (nodeType(pNode)) {

×

819

      case QUERY_NODE_VALUE:

×

820

        break;

×

821

      case QUERY_NODE_COLUMN:

×

824

        SExprNode* expNode = (SExprNode*)pNode;

×

825

        code = createResultData(&expNode->resType, rows, &output);

×

826

        if (code != TSDB_CODE_SUCCESS) {

×

827

          goto end;

×

829

        break;

×

832

      default:

×

833

        code = TSDB_CODE_OPS_NOT_SUPPORT;

×

834

        goto end;

×

837

    if (nodeType(pNode) == QUERY_NODE_COLUMN) {

×

838

      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;

×

839

      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, pSColumnNode->slotId);

×

840

      QUERY_CHECK_NULL(pColInfo, code, lino, end, terrno);

×

841

      code = colDataAssign(output.columnData, pColInfo, rows, NULL);

×

842

    } else if (nodeType(pNode) == QUERY_NODE_VALUE) {

×

843

      continue;

×

845

      code = scalarCalculate(pNode, pBlockList, &output);

×

848

    if (code != TSDB_CODE_SUCCESS) {

×

849

      releaseColInfoData(output.columnData);

×

850

      goto end;

×

853

    void* tmp = taosArrayPush(groupData, &output.columnData);

×

854

    QUERY_CHECK_NULL(tmp, code, lino, end, terrno);

×

857

  int32_t keyLen = 0;

×

859

  FOREACH(node, group) {

×

860

    SExprNode* pExpr = (SExprNode*)node;

×

861

    keyLen += pExpr->resType.bytes;

×

864

  int32_t nullFlagSize = sizeof(int8_t) * LIST_LENGTH(group);

×

865

  keyLen += nullFlagSize;

×

867

  keyBuf = taosMemoryCalloc(1, keyLen);

×

868

  if (keyBuf == NULL) {

×

869

    code = terrno;

×

870

    goto end;

×

873

  if (initRemainGroups) {

×

874

    pTableListInfo->remainGroups =

×

875

        taosHashInit(rows, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

×

876

    if (pTableListInfo->remainGroups == NULL) {

×

877

      code = terrno;

×

878

      goto end;

×

882

  for (int i = 0; i < rows; i++) {

×

883

    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);

×

884

    QUERY_CHECK_NULL(info, code, lino, end, terrno);

×

886

    char* isNull = (char*)keyBuf;

×

887

    char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(group);

×

888

    for (int j = 0; j < taosArrayGetSize(groupData); j++) {

×

889

      SColumnInfoData* pValue = (SColumnInfoData*)taosArrayGetP(groupData, j);

×

891

      if (colDataIsNull_s(pValue, i)) {

×

892

        isNull[j] = 1;

×

894

        isNull[j] = 0;

×

895

        char* data = colDataGetData(pValue, i);

×

896

        if (pValue->info.type == TSDB_DATA_TYPE_JSON) {

×

897

          if (tTagIsJson(data)) {

×

898

            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;

×

899

            goto end;

×

901

          if (tTagIsJsonNull(data)) {

×

902

            isNull[j] = 1;

×

903

            continue;

×

905

          int32_t len = getJsonValueLen(data);

×

906

          memcpy(pStart, data, len);

×

907

          pStart += len;

×

908

        } else if (IS_VAR_DATA_TYPE(pValue->info.type)) {

×

909

          if (varDataTLen(data) > pValue->info.bytes) {

×

910

            code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;

×

911

            goto end;

×

913

          memcpy(pStart, data, varDataTLen(data));

×

914

          pStart += varDataTLen(data);

×

916

          memcpy(pStart, data, pValue->info.bytes);

×

917

          pStart += pValue->info.bytes;

×

922

    int32_t len = (int32_t)(pStart - (char*)keyBuf);

×

923

    info->groupId = calcGroupId(keyBuf, len);

×

924

    if (initRemainGroups) {

×

926

      code = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId), &(info->uid),

×

928

      if (code == TSDB_CODE_DUP_KEY) {

×

929

        code = TSDB_CODE_SUCCESS;

×

931

      QUERY_CHECK_CODE(code, lino, end);

×

935

  if (tsTagFilterCache) {

×

936

    tableList = taosArrayDup(pTableListInfo->pTableList, NULL);

×

937

    QUERY_CHECK_NULL(tableList, code, lino, end, terrno);

×

939

    code = pAPI->metaFn.metaPutTbGroupToCache(pVnode, pTableListInfo->idInfo.suid, context.digest,

×

941

                                              taosArrayGetSize(tableList) * sizeof(STableKeyInfo));

×

942

    QUERY_CHECK_CODE(code, lino, end);

×

948

end:

×

949

  taosMemoryFreeClear(keyBuf);

×

950

  taosHashCleanup(ctx.colHash);

×

951

  taosArrayDestroy(ctx.cInfoList);

×

952

  blockDataDestroy(pResBlock);

×

953

  taosArrayDestroy(pBlockList);

×

954

  taosArrayDestroyEx(pUidTagList, freeItem);

×

955

  taosArrayDestroyP(groupData, releaseColInfoData);

×

956

  if (code != TSDB_CODE_SUCCESS) {

×

957

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

959

  return code;

×

962

static int32_t nameComparFn(const void* p1, const void* p2) {

×

963

  const char* pName1 = *(const char**)p1;

×

964

  const char* pName2 = *(const char**)p2;

×

966

  int32_t ret = strcmp(pName1, pName2);

×

967

  if (ret == 0) {

×

968

    return 0;

×

970

    return (ret > 0) ? 1 : -1;

×

974

static SArray* getTableNameList(const SNodeListNode* pList) {

×

975

  int32_t    code = TSDB_CODE_SUCCESS;

×

976

  int32_t    lino = 0;

×

977

  int32_t    len = LIST_LENGTH(pList->pNodeList);

×

978

  SListCell* cell = pList->pNodeList->pHead;

×

980

  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);

×

981

  QUERY_CHECK_NULL(pTbList, code, lino, _end, terrno);

×

983

  for (int i = 0; i < pList->pNodeList->length; i++) {

×

984

    SValueNode* valueNode = (SValueNode*)cell->pNode;

×

985

    if (!IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {

×

986

      terrno = TSDB_CODE_INVALID_PARA;

×

987

      taosArrayDestroy(pTbList);

×

988

      return NULL;

×

991

    char* name = varDataVal(valueNode->datum.p);

×

992

    void* tmp = taosArrayPush(pTbList, &name);

×

993

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

994

    cell = cell->pNext;

×

997

  size_t numOfTables = taosArrayGetSize(pTbList);

×

1000

  taosArraySort(pTbList, nameComparFn);

×

1003

  SArray* pNewList = taosArrayInit(taosArrayGetSize(pTbList), sizeof(void*));

×

1004

  QUERY_CHECK_NULL(pNewList, code, lino, _end, terrno);

×

1005

  void*   tmpTbl = taosArrayGet(pTbList, 0);

×

1006

  QUERY_CHECK_NULL(tmpTbl, code, lino, _end, terrno);

×

1007

  void*   tmp = taosArrayPush(pNewList, tmpTbl);

×

1008

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1010

  for (int32_t i = 1; i < numOfTables; ++i) {

×

1011

    char** name = taosArrayGetLast(pNewList);

×

1012

    char** nameInOldList = taosArrayGet(pTbList, i);

×

1013

    QUERY_CHECK_NULL(nameInOldList, code, lino, _end, terrno);

×

1014

    if (strcmp(*name, *nameInOldList) == 0) {

×

1015

      continue;

×

1018

    tmp = taosArrayPush(pNewList, nameInOldList);

×

1019

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1022

_end:

×

1023

  taosArrayDestroy(pTbList);

×

1024

  if (code != TSDB_CODE_SUCCESS) {

×

1025

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1026

    return NULL;

×

1028

  return pNewList;

×

1031

static int tableUidCompare(const void* a, const void* b) {

×

1032

  uint64_t u1 = *(uint64_t*)a;

×

1033

  uint64_t u2 = *(uint64_t*)b;

×

1035

  if (u1 == u2) {

×

1036

    return 0;

×

1039

  return u1 < u2 ? -1 : 1;

×

1042

static int32_t filterTableInfoCompare(const void* a, const void* b) {

×

1043

  STUidTagInfo* p1 = (STUidTagInfo*)a;

×

1044

  STUidTagInfo* p2 = (STUidTagInfo*)b;

×

1046

  if (p1->uid == p2->uid) {

×

1047

    return 0;

×

1050

  return p1->uid < p2->uid ? -1 : 1;

×

1053

static FilterCondType checkTagCond(SNode* cond) {

×

1054

  if (nodeType(cond) == QUERY_NODE_OPERATOR) {

×

1055

    return FILTER_NO_LOGIC;

×

1057

  if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {

×

1058

    return FILTER_AND;

×

1060

  return FILTER_OTHER;

×

1063

static int32_t optimizeTbnameInCond(void* pVnode, int64_t suid, SArray* list, SNode* cond, SStorageAPI* pAPI) {

×

1064

  int32_t ret = -1;

×

1065

  int32_t ntype = nodeType(cond);

×

1067

  if (ntype == QUERY_NODE_OPERATOR) {

×

1068

    ret = optimizeTbnameInCondImpl(pVnode, list, cond, pAPI, suid);

×

1071

  if (ntype != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {

×

1072

    return ret;

×

1075

  bool                 hasTbnameCond = false;

×

1076

  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;

×

1077

  SNodeList*           pList = (SNodeList*)pNode->pParameterList;

×

1079

  int32_t len = LIST_LENGTH(pList);

×

1080

  if (len <= 0) {

×

1081

    return ret;

×

1084

  SListCell* cell = pList->pHead;

×

1085

  for (int i = 0; i < len; i++) {

×

1086

    if (cell == NULL) break;

×

1087

    if (optimizeTbnameInCondImpl(pVnode, list, cell->pNode, pAPI, suid) == 0) {

×

1088

      hasTbnameCond = true;

×

1089

      break;

×

1091

    cell = cell->pNext;

×

1094

  taosArraySort(list, filterTableInfoCompare);

×

1095

  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);

×

1097

  if (hasTbnameCond) {

×

1098

    ret = pAPI->metaFn.getTableTagsByUid(pVnode, suid, list);

×

1101

  return ret;

×

1105

static int32_t optimizeTbnameInCondImpl(void* pVnode, SArray* pExistedUidList, SNode* pTagCond,

×

1107

  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {

×

1108

    return -1;

×

1111

  SOperatorNode* pNode = (SOperatorNode*)pTagCond;

×

1112

  if (pNode->opType != OP_TYPE_IN) {

×

1113

    return -1;

×

1116

  if ((pNode->pLeft != NULL && nodeType(pNode->pLeft) == QUERY_NODE_COLUMN &&

×

1117

       ((SColumnNode*)pNode->pLeft)->colType == COLUMN_TYPE_TBNAME) &&

×

1118

      (pNode->pRight != NULL && nodeType(pNode->pRight) == QUERY_NODE_NODE_LIST)) {

×

1119

    SNodeListNode* pList = (SNodeListNode*)pNode->pRight;

×

1121

    int32_t len = LIST_LENGTH(pList->pNodeList);

×

1122

    if (len <= 0) {

×

1123

      return -1;

×

1126

    SArray*   pTbList = getTableNameList(pList);

×

1127

    int32_t   numOfTables = taosArrayGetSize(pTbList);

×

1128

    SHashObj* uHash = NULL;

×

1130

    size_t numOfExisted = taosArrayGetSize(pExistedUidList);  // len > 0 means there already have uids

×

1131

    if (numOfExisted > 0) {

×

1132

      uHash = taosHashInit(numOfExisted / 0.7, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

×

1133

      if (!uHash) {

×

1134

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1135

        return terrno;

×

1138

      for (int i = 0; i < numOfExisted; i++) {

×

1139

        STUidTagInfo* pTInfo = taosArrayGet(pExistedUidList, i);

×

1140

        if (!pTInfo) {

×

1141

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1142

          return terrno;

×

1144

        int32_t       tempRes = taosHashPut(uHash, &pTInfo->uid, sizeof(uint64_t), &i, sizeof(i));

×

1145

        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {

×

1146

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));

×

1147

          return tempRes;

×

1152

    for (int i = 0; i < numOfTables; i++) {

×

1153

      char* name = taosArrayGetP(pTbList, i);

×

1155

      uint64_t uid = 0, csuid = 0;

×

1156

      if (pStoreAPI->metaFn.getTableUidByName(pVnode, name, &uid) == 0) {

×

1157

        ETableType tbType = TSDB_TABLE_MAX;

×

1158

        if (pStoreAPI->metaFn.getTableTypeSuidByName(pVnode, name, &tbType, &csuid) == 0 && tbType == TSDB_CHILD_TABLE) {

×

1159

          if (suid != csuid) {

×

1160

            continue;

×

1162

          if (NULL == uHash || taosHashGet(uHash, &uid, sizeof(uid)) == NULL) {

×

1163

            STUidTagInfo s = {.uid = uid, .name = name, .pTagVal = NULL};

×

1164

            void*        tmp = taosArrayPush(pExistedUidList, &s);

×

1165

            if (!tmp) {

×

1166

              return terrno;

×

1170

          taosArrayDestroy(pTbList);

×

1171

          taosHashCleanup(uHash);

×

1172

          return -1;

×

1176

        terrno = 0;

×

1180

    taosHashCleanup(uHash);

×

1181

    taosArrayDestroy(pTbList);

×

1182

    return 0;

×

1185

  return -1;

×

1188

SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, SArray* pUidTagList, void* pVnode,

×

1190

  int32_t      code = TSDB_CODE_SUCCESS;

×

1191

  int32_t      lino = 0;

×

1192

  SSDataBlock* pResBlock = NULL;

×

1193

  code = createDataBlock(&pResBlock);

×

1194

  QUERY_CHECK_CODE(code, lino, _end);

×

1196

  for (int32_t i = 0; i < taosArrayGetSize(pColList); ++i) {

×

1197

    SColumnInfoData colInfo = {0};

×

1198

    void* tmp = taosArrayGet(pColList, i);

×

1199

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1200

    colInfo.info = *(SColumnInfo*)tmp;

×

1201

    code = blockDataAppendColInfo(pResBlock, &colInfo);

×

1202

    QUERY_CHECK_CODE(code, lino, _end);

×

1205

  code = blockDataEnsureCapacity(pResBlock, numOfTables);

×

1206

  if (code != TSDB_CODE_SUCCESS) {

×

1207

    terrno = code;

×

1208

    blockDataDestroy(pResBlock);

×

1209

    return NULL;

×

1212

  pResBlock->info.rows = numOfTables;

×

1214

  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);

×

1216

  for (int32_t i = 0; i < numOfTables; i++) {

×

1217

    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);

×

1218

    QUERY_CHECK_NULL(p1, code, lino, _end, terrno);

×

1220

    for (int32_t j = 0; j < numOfCols; j++) {

×

1221

      SColumnInfoData* pColInfo = (SColumnInfoData*)taosArrayGet(pResBlock->pDataBlock, j);

×

1222

      QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);

×

1224

      if (pColInfo->info.colId == -1) {  // tbname

×

1225

        char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

1226

        if (p1->name != NULL) {

×

1227

          STR_TO_VARSTR(str, p1->name);

×

1229

          code = pStorageAPI->metaFn.getTableNameByUid(pVnode, p1->uid, str);

×

1230

          QUERY_CHECK_CODE(code, lino, _end);

×

1233

        code = colDataSetVal(pColInfo, i, str, false);

×

1234

        QUERY_CHECK_CODE(code, lino, _end);

×

1239

        STagVal tagVal = {0};

×

1240

        tagVal.cid = pColInfo->info.colId;

×

1241

        if (p1->pTagVal == NULL) {

×

1242

          colDataSetNULL(pColInfo, i);

×

1244

          const char* p = pStorageAPI->metaFn.extractTagVal(p1->pTagVal, pColInfo->info.type, &tagVal);

×

1246

          if (p == NULL || (pColInfo->info.type == TSDB_DATA_TYPE_JSON && ((STag*)p)->nTag == 0)) {

×

1247

            colDataSetNULL(pColInfo, i);

×

1248

          } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) {

×

1249

            code = colDataSetVal(pColInfo, i, p, false);

×

1250

            QUERY_CHECK_CODE(code, lino, _end);

×

1251

          } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {

×

1252

            char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1);

×

1253

            QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1254

            varDataSetLen(tmp, tagVal.nData);

×

1255

            memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);

×

1256

            code = colDataSetVal(pColInfo, i, tmp, false);

×

1260

            taosMemoryFree(tmp);

×

1261

            QUERY_CHECK_CODE(code, lino, _end);

×

1263

            code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);

×

1264

            QUERY_CHECK_CODE(code, lino, _end);

×

1278

_end:

×

1279

  if (code != TSDB_CODE_SUCCESS) {

×

1280

    blockDataDestroy(pResBlock);

×

1281

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1282

    terrno = code;

×

1283

    return NULL;

×

1285

  return pResBlock;

×

1288

static int32_t doSetQualifiedUid(STableListInfo* pListInfo, SArray* pUidList, const SArray* pUidTagList,

×

1290

  taosArrayClear(pUidList);

×

1292

  STableKeyInfo info = {.uid = 0, .groupId = 0};

×

1293

  int32_t       numOfTables = taosArrayGetSize(pUidTagList);

×

1294

  for (int32_t i = 0; i < numOfTables; ++i) {

×

1295

    if (pResultList[i]) {

×

1296

      STUidTagInfo* tmpTag = (STUidTagInfo*)taosArrayGet(pUidTagList, i);

×

1297

      if (!tmpTag) {

×

1298

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1299

        return terrno;

×

1301

      uint64_t uid = tmpTag->uid;

×

1302

      qDebug("tagfilter get uid:%" PRId64 ", res:%d", uid, pResultList[i]);

×

1304

      info.uid = uid;

×

1305

      void* p = taosArrayPush(pListInfo->pTableList, &info);

×

1306

      if (p == NULL) {

×

1307

        return terrno;

×

1310

      if (addUid) {

×

1311

        void* tmp = taosArrayPush(pUidList, &uid);

×

1312

        if (tmp == NULL) {

×

1313

          return terrno;

×

1319

  return TSDB_CODE_SUCCESS;

×

1322

static int32_t copyExistedUids(SArray* pUidTagList, const SArray* pUidList) {

×

1323

  int32_t code = TSDB_CODE_SUCCESS;

×

1324

  int32_t numOfExisted = taosArrayGetSize(pUidList);

×

1325

  if (numOfExisted == 0) {

×

1326

    return code;

×

1329

  for (int32_t i = 0; i < numOfExisted; ++i) {

×

1330

    uint64_t*    uid = taosArrayGet(pUidList, i);

×

1331

    if (!uid) {

×

1332

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1333

      return terrno;

×

1335

    STUidTagInfo info = {.uid = *uid};

×

1336

    void*        tmp = taosArrayPush(pUidTagList, &info);

×

1337

    if (!tmp) {

×

1338

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1339

      return code;

×

1342

  return code;

×

1345

static int32_t doFilterByTagCond(STableListInfo* pListInfo, SArray* pUidList, SNode* pTagCond, void* pVnode,

×

1347

  *listAdded = false;

×

1348

  if (pTagCond == NULL) {

×

1349

    return TSDB_CODE_SUCCESS;

×

1352

  terrno = TSDB_CODE_SUCCESS;

×

1354

  int32_t      lino = 0;

×

1355

  int32_t      code = TSDB_CODE_SUCCESS;

×

1356

  SArray*      pBlockList = NULL;

×

1357

  SSDataBlock* pResBlock = NULL;

×

1358

  SScalarParam output = {0};

×

1359

  SArray*      pUidTagList = NULL;

×

1361

  tagFilterAssist ctx = {0};

×

1362

  ctx.colHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK);

×

1363

  if (ctx.colHash == NULL) {

×

1364

    code = terrno;

×

1365

    QUERY_CHECK_CODE(code, lino, end);

×

1368

  ctx.cInfoList = taosArrayInit(4, sizeof(SColumnInfo));

×

1369

  if (ctx.cInfoList == NULL) {

×

1370

    code = terrno;

×

1371

    QUERY_CHECK_CODE(code, lino, end);

×

1374

  nodesRewriteExprPostOrder(&pTagCond, getColumn, (void*)&ctx);

×

1375

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

1376

    terrno = code = ctx.code;

×

1377

    goto end;

×

1380

  SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};

×

1383

  pUidTagList = taosArrayInit(10, sizeof(STUidTagInfo));

×

1384

  QUERY_CHECK_NULL(pUidTagList, code, lino, end, terrno);

×

1386

  code = copyExistedUids(pUidTagList, pUidList);

×

1387

  QUERY_CHECK_CODE(code, lino, end);

×

1389

  FilterCondType condType = checkTagCond(pTagCond);

×

1391

  int32_t filter = optimizeTbnameInCond(pVnode, pListInfo->idInfo.suid, pUidTagList, pTagCond, pAPI);

×

1392

  if (filter == 0) {  // tbname in filter is activated, do nothing and return

×

1393

    taosArrayClear(pUidList);

×

1395

    int32_t numOfRows = taosArrayGetSize(pUidTagList);

×

1396

    code = taosArrayEnsureCap(pUidList, numOfRows);

×

1397

    QUERY_CHECK_CODE(code, lino, end);

×

1399

    for (int32_t i = 0; i < numOfRows; ++i) {

×

1400

      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);

×

1401

      QUERY_CHECK_NULL(pInfo, code, lino, end, terrno);

×

1402

      void*         tmp = taosArrayPush(pUidList, &pInfo->uid);

×

1403

      QUERY_CHECK_NULL(tmp, code, lino, end, terrno);

×

1405

    terrno = 0;

×

1407

    if ((condType == FILTER_NO_LOGIC || condType == FILTER_AND) && status != SFLT_NOT_INDEX) {

×

1408

      code = pAPI->metaFn.getTableTagsByUid(pVnode, pListInfo->idInfo.suid, pUidTagList);

×

1410

      code = pAPI->metaFn.getTableTags(pVnode, pListInfo->idInfo.suid, pUidTagList);

×

1412

    if (code != TSDB_CODE_SUCCESS) {

×

1413

      qError("failed to get table tags from meta, reason:%s, suid:%" PRIu64, tstrerror(code), pListInfo->idInfo.suid);

×

1414

      terrno = code;

×

1415

      QUERY_CHECK_CODE(code, lino, end);

×

1419

  int32_t numOfTables = taosArrayGetSize(pUidTagList);

×

1420

  if (numOfTables == 0) {

×

1421

    goto end;

×

1424

  pResBlock = createTagValBlockForFilter(ctx.cInfoList, numOfTables, pUidTagList, pVnode, pAPI);

×

1425

  if (pResBlock == NULL) {

×

1426

    code = terrno;

×

1427

    QUERY_CHECK_CODE(code, lino, end);

×

1432

  pBlockList = taosArrayInit(2, POINTER_BYTES);

×

1433

  QUERY_CHECK_NULL(pBlockList, code, lino, end, terrno);

×

1435

  void* tmp = taosArrayPush(pBlockList, &pResBlock);

×

1436

  QUERY_CHECK_NULL(tmp, code, lino, end, terrno);

×

1438

  code = createResultData(&type, numOfTables, &output);

×

1439

  if (code != TSDB_CODE_SUCCESS) {

×

1440

    terrno = code;

×

1441

    QUERY_CHECK_CODE(code, lino, end);

×

1444

  code = scalarCalculate(pTagCond, pBlockList, &output);

×

1445

  if (code != TSDB_CODE_SUCCESS) {

×

1446

    qError("failed to calculate scalar, reason:%s", tstrerror(code));

×

1447

    terrno = code;

×

1448

    QUERY_CHECK_CODE(code, lino, end);

×

1451

  code = doSetQualifiedUid(pListInfo, pUidList, pUidTagList, (bool*)output.columnData->pData, addUid);

×

1452

  if (code != TSDB_CODE_SUCCESS) {

×

1453

    terrno = code;

×

1454

    QUERY_CHECK_CODE(code, lino, end);

×

1456

  *listAdded = true;

×

1458

end:

×

1459

  if (code != TSDB_CODE_SUCCESS) {

×

1460

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1462

  taosHashCleanup(ctx.colHash);

×

1463

  taosArrayDestroy(ctx.cInfoList);

×

1464

  blockDataDestroy(pResBlock);

×

1465

  taosArrayDestroy(pBlockList);

×

1466

  taosArrayDestroyEx(pUidTagList, freeItem);

×

1468

  colDataDestroy(output.columnData);

×

1469

  taosMemoryFreeClear(output.columnData);

×

1470

  return code;

×

1473

int32_t getTableList(void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagCond, SNode* pTagIndexCond,

×

1475

  int32_t code = TSDB_CODE_SUCCESS;

×

1476

  int32_t lino = 0;

×

1477

  size_t  numOfTables = 0;

×

1478

  bool    listAdded = false;

×

1480

  pListInfo->idInfo.suid = pScanNode->suid;

×

1481

  pListInfo->idInfo.tableType = pScanNode->tableType;

×

1483

  SArray* pUidList = taosArrayInit(8, sizeof(uint64_t));

×

1484

  QUERY_CHECK_NULL(pUidList, code, lino, _error, terrno);

×

1486

  SIdxFltStatus status = SFLT_NOT_INDEX;

×

1487

  if (pScanNode->tableType != TSDB_SUPER_TABLE) {

×

1488

    pListInfo->idInfo.uid = pScanNode->uid;

×

1489

    if (pStorageAPI->metaFn.isTableExisted(pVnode, pScanNode->uid)) {

×

1490

      void* tmp = taosArrayPush(pUidList, &pScanNode->uid);

×

1491

      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);

×

1493

    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, false, &listAdded);

×

1494

    QUERY_CHECK_CODE(code, lino, _end);

×

1496

    T_MD5_CTX context = {0};

×

1498

    if (tsTagFilterCache) {

×

1500

      code = genTagFilterDigest(pTagCond, &context);

×

1501

      QUERY_CHECK_CODE(code, lino, _error);

×

1503

      bool acquired = false;

×

1504

      code = pStorageAPI->metaFn.getCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),

×

1506

      QUERY_CHECK_CODE(code, lino, _error);

×

1508

      if (acquired) {

×

1509

        digest[0] = 1;

×

1510

        memcpy(digest + 1, context.digest, tListLen(context.digest));

×

1511

        qDebug("retrieve table uid list from cache, numOfTables:%d", (int32_t)taosArrayGetSize(pUidList));

×

1512

        goto _end;

×

1516

    if (!pTagCond) {  // no tag filter condition exists, let's fetch all tables of this super table

×

1517

      code = pStorageAPI->metaFn.getChildTableList(pVnode, pScanNode->suid, pUidList);

×

1518

      QUERY_CHECK_CODE(code, lino, _error);

×

1521

      if (pTagIndexCond) {

×

1522

        void* pIndex = pStorageAPI->metaFn.getInvertIndex(pVnode);

×

1524

        SIndexMetaArg metaArg = {.metaEx = pVnode,

×

1525

                                 .idx = pStorageAPI->metaFn.storeGetIndexInfo(pVnode),

×

1527

                                 .suid = pScanNode->uid};

×

1529

        status = SFLT_NOT_INDEX;

×

1530

        code = doFilterTag(pTagIndexCond, &metaArg, pUidList, &status, &pStorageAPI->metaFilter);

×

1531

        if (code != 0 || status == SFLT_NOT_INDEX) {  // temporarily disable it for performance sake

×

1532

          qDebug("failed to get tableIds from index, suid:%" PRIu64, pScanNode->uid);

×

1534

          qDebug("succ to get filter result, table num: %d", (int)taosArrayGetSize(pUidList));

×

1539

    code = doFilterByTagCond(pListInfo, pUidList, pTagCond, pVnode, status, pStorageAPI, tsTagFilterCache, &listAdded);

×

1540

    QUERY_CHECK_CODE(code, lino, _end);

×

1543

    numOfTables = taosArrayGetSize(pUidList);

×

1545

    if (tsTagFilterCache) {

×

1546

      size_t size = numOfTables * sizeof(uint64_t) + sizeof(int32_t);

×

1547

      char*  pPayload = taosMemoryMalloc(size);

×

1548

      QUERY_CHECK_NULL(pPayload, code, lino, _end, terrno);

×

1550

      *(int32_t*)pPayload = numOfTables;

×

1551

      if (numOfTables > 0) {

×

1552

        void* tmp = taosArrayGet(pUidList, 0);

×

1553

        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1554

        memcpy(pPayload + sizeof(int32_t), tmp, numOfTables * sizeof(uint64_t));

×

1557

      code = pStorageAPI->metaFn.putCachedTableList(pVnode, pScanNode->suid, context.digest, tListLen(context.digest),

×

1559

      QUERY_CHECK_CODE(code, lino, _error);

×

1561

      digest[0] = 1;

×

1562

      memcpy(digest + 1, context.digest, tListLen(context.digest));

×

1566

_end:

×

1567

  if (!listAdded) {

×

1568

    numOfTables = taosArrayGetSize(pUidList);

×

1569

    for (int i = 0; i < numOfTables; i++) {

×

1570

      void* tmp = taosArrayGet(pUidList, i);

×

1571

      QUERY_CHECK_NULL(tmp, code, lino, _error, terrno);

×

1572

      STableKeyInfo info = {.uid = *(uint64_t*)tmp, .groupId = 0};

×

1574

      void* p = taosArrayPush(pListInfo->pTableList, &info);

×

1575

      if (p == NULL) {

×

1576

        taosArrayDestroy(pUidList);

×

1577

        return terrno;

×

1580

      qTrace("tagfilter get uid:%" PRIu64 ", %s", info.uid, idstr);

×

1584

_error:

×

1585

  taosArrayDestroy(pUidList);

×

1586

  if (code != TSDB_CODE_SUCCESS) {

×

1587

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1589

  return code;

×

1592

int32_t qGetTableList(int64_t suid, void* pVnode, void* node, SArray** tableList, void* pTaskInfo) {

×

1593

  int32_t        code = TSDB_CODE_SUCCESS;

×

1594

  int32_t        lino = 0;

×

1595

  SSubplan*      pSubplan = (SSubplan*)node;

×

1596

  SScanPhysiNode pNode = {0};

×

1597

  pNode.suid = suid;

×

1598

  pNode.uid = suid;

×

1599

  pNode.tableType = TSDB_SUPER_TABLE;

×

1601

  STableListInfo* pTableListInfo = tableListCreate();

×

1602

  QUERY_CHECK_NULL(pTableListInfo, code, lino, _end, terrno);

×

1603

  uint8_t         digest[17] = {0};

×

1605

      getTableList(pVnode, &pNode, pSubplan ? pSubplan->pTagCond : NULL, pSubplan ? pSubplan->pTagIndexCond : NULL,

×

1607

  QUERY_CHECK_CODE(code, lino, _end);

×

1608

  *tableList = pTableListInfo->pTableList;

×

1609

  pTableListInfo->pTableList = NULL;

×

1610

  tableListDestroy(pTableListInfo);

×

1612

_end:

×

1613

  if (code != TSDB_CODE_SUCCESS) {

×

1614

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1616

  return code;

×

1619

size_t getTableTagsBufLen(const SNodeList* pGroups) {

×

1620

  size_t keyLen = 0;

×

1623

  FOREACH(node, pGroups) {

×

1624

    SExprNode* pExpr = (SExprNode*)node;

×

1625

    keyLen += pExpr->resType.bytes;

×

1628

  keyLen += sizeof(int8_t) * LIST_LENGTH(pGroups);

×

1629

  return keyLen;

×

1632

int32_t getGroupIdFromTagsVal(void* pVnode, uint64_t uid, SNodeList* pGroupNode, char* keyBuf, uint64_t* pGroupId,

×

1634

  SMetaReader mr = {0};

×

1636

  pAPI->metaReaderFn.initReader(&mr, pVnode, META_READER_LOCK, &pAPI->metaFn);

×

1637

  if (pAPI->metaReaderFn.getEntryGetUidCache(&mr, uid) != 0) {  // table not exist

×

1638

    pAPI->metaReaderFn.clearReader(&mr);

×

1639

    return TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

1642

  SNodeList* groupNew = NULL;

×

1643

  int32_t    code = nodesCloneList(pGroupNode, &groupNew);

×

1644

  if (TSDB_CODE_SUCCESS != code) {

×

1645

    pAPI->metaReaderFn.clearReader(&mr);

×

1646

    return code;

×

1649

  STransTagExprCtx ctx = {.code = 0, .pReader = &mr};

×

1650

  nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);

×

1651

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

1652

    nodesDestroyList(groupNew);

×

1653

    pAPI->metaReaderFn.clearReader(&mr);

×

1654

    return code;

×

1656

  char* isNull = (char*)keyBuf;

×

1657

  char* pStart = (char*)keyBuf + sizeof(int8_t) * LIST_LENGTH(pGroupNode);

×

1660

  int32_t index = 0;

×

1661

  FOREACH(pNode, groupNew) {

×

1662

    SNode*  pNew = NULL;

×

1663

    int32_t code = scalarCalculateConstants(pNode, &pNew);

×

1664

    if (TSDB_CODE_SUCCESS == code) {

×

1665

      REPLACE_NODE(pNew);

×

1667

      nodesDestroyList(groupNew);

×

1668

      pAPI->metaReaderFn.clearReader(&mr);

×

1669

      return code;

×

1672

    if (nodeType(pNew) != QUERY_NODE_VALUE) {

×

1673

      nodesDestroyList(groupNew);

×

1674

      pAPI->metaReaderFn.clearReader(&mr);

×

1675

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));

×

1676

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1678

    SValueNode* pValue = (SValueNode*)pNew;

×

1680

    if (pValue->node.resType.type == TSDB_DATA_TYPE_NULL || pValue->isNull) {

×

1681

      isNull[index++] = 1;

×

1682

      continue;

×

1684

      isNull[index++] = 0;

×

1685

      char* data = nodesGetValueFromNode(pValue);

×

1686

      if (pValue->node.resType.type == TSDB_DATA_TYPE_JSON) {

×

1687

        if (tTagIsJson(data)) {

×

1688

          terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;

×

1689

          nodesDestroyList(groupNew);

×

1690

          pAPI->metaReaderFn.clearReader(&mr);

×

1691

          return terrno;

×

1693

        int32_t len = getJsonValueLen(data);

×

1694

        memcpy(pStart, data, len);

×

1695

        pStart += len;

×

1696

      } else if (IS_VAR_DATA_TYPE(pValue->node.resType.type)) {

×

1697

        memcpy(pStart, data, varDataTLen(data));

×

1698

        pStart += varDataTLen(data);

×

1700

        memcpy(pStart, data, pValue->node.resType.bytes);

×

1701

        pStart += pValue->node.resType.bytes;

×

1706

  int32_t len = (int32_t)(pStart - (char*)keyBuf);

×

1707

  *pGroupId = calcGroupId(keyBuf, len);

×

1709

  nodesDestroyList(groupNew);

×

1710

  pAPI->metaReaderFn.clearReader(&mr);

×

1712

  return TSDB_CODE_SUCCESS;

×

1715

SArray* makeColumnArrayFromList(SNodeList* pNodeList) {

×

1716

  if (!pNodeList) {

×

1717

    return NULL;

×

1720

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

1721

  SArray* pList = taosArrayInit(numOfCols, sizeof(SColumn));

×

1722

  if (pList == NULL) {

×

1723

    return NULL;

×

1726

  for (int32_t i = 0; i < numOfCols; ++i) {

×

1727

    SColumnNode* pColNode = (SColumnNode*)nodesListGetNode(pNodeList, i);

×

1728

    if (!pColNode) {

×

1729

      taosArrayDestroy(pList);

×

1730

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));

×

1731

      return NULL;

×

1735

    SColumn c = {0};

×

1736

    c.slotId = pColNode->slotId;

×

1737

    c.colId = pColNode->colId;

×

1738

    c.type = pColNode->node.resType.type;

×

1739

    c.bytes = pColNode->node.resType.bytes;

×

1740

    c.precision = pColNode->node.resType.precision;

×

1741

    c.scale = pColNode->node.resType.scale;

×

1743

    void* tmp = taosArrayPush(pList, &c);

×

1744

    if (!tmp) {

×

1745

      taosArrayDestroy(pList);

×

1746

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

1747

      return NULL;

×

1751

  return pList;

×

1754

int32_t extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols,

×

1756

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

1757

  int32_t code = TSDB_CODE_SUCCESS;

×

1758

  int32_t lino = 0;

×

1760

  pMatchInfo->matchType = type;

×

1762

  SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchItem));

×

1763

  if (pList == NULL) {

×

1764

    code = terrno;

×

1765

    return code;

×

1768

  for (int32_t i = 0; i < numOfCols; ++i) {

×

1769

    STargetNode* pNode = (STargetNode*)nodesListGetNode(pNodeList, i);

×

1770

    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);

×

1771

    if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN) {

×

1772

      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;

×

1774

      SColMatchItem c = {.needOutput = true};

×

1775

      c.colId = pColNode->colId;

×

1776

      c.srcSlotId = pColNode->slotId;

×

1777

      c.dstSlotId = pNode->slotId;

×

1778

      c.isPk = pColNode->isPk;

×

1779

      c.dataType = pColNode->node.resType;

×

1780

      void* tmp = taosArrayPush(pList, &c);

×

1781

      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

1786

  *numOfOutputCols = 0;

×

1787

  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);

×

1788

  for (int32_t i = 0; i < num; ++i) {

×

1789

    SSlotDescNode* pNode = (SSlotDescNode*)nodesListGetNode(pOutputNodeList->pSlots, i);

×

1790

    QUERY_CHECK_NULL(pNode, code, lino, _end, terrno);

×

1794

    if (pNode->slotId >= numOfCols) {

×

1795

      (*numOfOutputCols) += 1;

×

1796

      continue;

×

1799

    SColMatchItem* info = NULL;

×

1800

    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {

×

1801

      info = taosArrayGet(pList, j);

×

1802

      QUERY_CHECK_NULL(info, code, lino, _end, terrno);

×

1803

      if (info->dstSlotId == pNode->slotId) {

×

1804

        break;

×

1808

    if (pNode->output) {

×

1809

      (*numOfOutputCols) += 1;

×

1810

    } else if (info != NULL) {

×

1812

      info->needOutput = false;

×

1816

  pMatchInfo->pList = pList;

×

1818

_end:

×

1819

  if (code != TSDB_CODE_SUCCESS) {

×

1820

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

1822

  return code;

×

1825

static SResSchema createResSchema(int32_t type, int32_t bytes, int32_t slotId, int32_t scale, int32_t precision,

×

1827

  SResSchema s = {0};

×

1828

  s.scale = scale;

×

1829

  s.type = type;

×

1830

  s.bytes = bytes;

×

1831

  s.slotId = slotId;

×

1832

  s.precision = precision;

×

1833

  tstrncpy(s.name, name, tListLen(s.name));

×

1835

  return s;

×

1838

static SColumn* createColumn(int32_t blockId, int32_t slotId, int32_t colId, SDataType* pType, EColumnType colType) {

×

1839

  SColumn* pCol = taosMemoryCalloc(1, sizeof(SColumn));

×

1840

  if (pCol == NULL) {

×

1841

    return NULL;

×

1844

  pCol->slotId = slotId;

×

1845

  pCol->colId = colId;

×

1846

  pCol->bytes = pType->bytes;

×

1847

  pCol->type = pType->type;

×

1848

  pCol->scale = pType->scale;

×

1849

  pCol->precision = pType->precision;

×

1850

  pCol->dataBlockId = blockId;

×

1851

  pCol->colType = colType;

×

1852

  return pCol;

×

1855

int32_t createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {

×

1856

  int32_t code = TSDB_CODE_SUCCESS;

×

1857

  int32_t lino = 0;

×

1858

  pExp->base.numOfParams = 0;

×

1859

  pExp->base.pParam = NULL;

×

1860

  pExp->pExpr = taosMemoryCalloc(1, sizeof(tExprNode));

×

1861

  QUERY_CHECK_NULL(pExp->pExpr, code, lino, _end, terrno);

×

1863

  pExp->pExpr->_function.num = 1;

×

1864

  pExp->pExpr->_function.functionId = -1;

×

1866

  int32_t type = nodeType(pNode);

×

1868

  if (type == QUERY_NODE_COLUMN) {

×

1869

    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;

×

1870

    SColumnNode* pColNode = (SColumnNode*)pNode;

×

1872

    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));

×

1873

    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);

×

1875

    pExp->base.numOfParams = 1;

×

1877

    SDataType* pType = &pColNode->node.resType;

×

1879

        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pColNode->colName);

×

1881

    pExp->base.pParam[0].pCol =

×

1882

        createColumn(pColNode->dataBlockId, pColNode->slotId, pColNode->colId, pType, pColNode->colType);

×

1883

    QUERY_CHECK_NULL(pExp->base.pParam[0].pCol, code, lino, _end, terrno);

×

1885

    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_COLUMN;

×

1886

  } else if (type == QUERY_NODE_VALUE) {

×

1887

    pExp->pExpr->nodeType = QUERY_NODE_VALUE;

×

1888

    SValueNode* pValNode = (SValueNode*)pNode;

×

1890

    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));

×

1891

    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);

×

1893

    pExp->base.numOfParams = 1;

×

1895

    SDataType* pType = &pValNode->node.resType;

×

1897

        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pValNode->node.aliasName);

×

1898

    pExp->base.pParam[0].type = FUNC_PARAM_TYPE_VALUE;

×

1899

    code = nodesValueNodeToVariant(pValNode, &pExp->base.pParam[0].param);

×

1900

    QUERY_CHECK_CODE(code, lino, _end);

×

1901

  } else if (type == QUERY_NODE_FUNCTION) {

×

1902

    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;

×

1903

    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;

×

1905

    SDataType* pType = &pFuncNode->node.resType;

×

1907

        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pFuncNode->node.aliasName);

×

1908

    tExprNode* pExprNode = pExp->pExpr;

×

1910

    pExprNode->_function.functionId = pFuncNode->funcId;

×

1911

    pExprNode->_function.pFunctNode = pFuncNode;

×

1912

    pExprNode->_function.functionType = pFuncNode->funcType;

×

1914

    tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName));

×

1918

    const char* name = "tbname";

×

1919

    int32_t     len = strlen(name);

×

1921

    if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&

×

1922

        pExprNode->_function.functionName[len] == 0) {

×

1923

      pFuncNode->pParameterList = NULL;

×

1924

      int32_t     code = nodesMakeList(&pFuncNode->pParameterList);

×

1925

      SValueNode* res = NULL;

×

1926

      if (TSDB_CODE_SUCCESS == code) {

×

1927

        code = nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&res);

×

1929

      QUERY_CHECK_CODE(code, lino, _end);

×

1930

      res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};

×

1931

      code = nodesListAppend(pFuncNode->pParameterList, (SNode*)res);

×

1932

      if (code != TSDB_CODE_SUCCESS) {

×

1933

        nodesDestroyNode((SNode*)res);

×

1934

        res = NULL;

×

1936

      QUERY_CHECK_CODE(code, lino, _end);

×

1940

    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);

×

1942

    pExp->base.pParam = taosMemoryCalloc(numOfParam, sizeof(SFunctParam));

×

1943

    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);

×

1944

    pExp->base.numOfParams = numOfParam;

×

1946

    for (int32_t j = 0; j < numOfParam && TSDB_CODE_SUCCESS == code; ++j) {

×

1947

      SNode* p1 = nodesListGetNode(pFuncNode->pParameterList, j);

×

1948

      QUERY_CHECK_NULL(p1, code, lino, _end, terrno);

×

1949

      if (p1->type == QUERY_NODE_COLUMN) {

×

1950

        SColumnNode* pcn = (SColumnNode*)p1;

×

1952

        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_COLUMN;

×

1953

        pExp->base.pParam[j].pCol =

×

1954

            createColumn(pcn->dataBlockId, pcn->slotId, pcn->colId, &pcn->node.resType, pcn->colType);

×

1955

        QUERY_CHECK_NULL(pExp->base.pParam[j].pCol, code, lino, _end, terrno);

×

1956

      } else if (p1->type == QUERY_NODE_VALUE) {

×

1957

        SValueNode* pvn = (SValueNode*)p1;

×

1958

        pExp->base.pParam[j].type = FUNC_PARAM_TYPE_VALUE;

×

1959

        code = nodesValueNodeToVariant(pvn, &pExp->base.pParam[j].param);

×

1960

        QUERY_CHECK_CODE(code, lino, _end);

×

1963

    pExp->pExpr->_function.bindExprID = ((SExprNode*)pNode)->bindExprID;

×

1964

  } else if (type == QUERY_NODE_OPERATOR) {

×

1965

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

1966

    SOperatorNode* pOpNode = (SOperatorNode*)pNode;

×

1968

    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));

×

1969

    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);

×

1970

    pExp->base.numOfParams = 1;

×

1972

    SDataType* pType = &pOpNode->node.resType;

×

1974

        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pOpNode->node.aliasName);

×

1975

    pExp->pExpr->_optrRoot.pRootNode = pNode;

×

1976

  } else if (type == QUERY_NODE_CASE_WHEN) {

×

1977

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

1978

    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;

×

1980

    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));

×

1981

    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);

×

1982

    pExp->base.numOfParams = 1;

×

1984

    SDataType* pType = &pCaseNode->node.resType;

×

1986

        createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCaseNode->node.aliasName);

×

1987

    pExp->pExpr->_optrRoot.pRootNode = pNode;

×

1988

  } else if (type == QUERY_NODE_LOGIC_CONDITION) {

×

1989

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

1990

    SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;

×

1991

    pExp->base.pParam = taosMemoryCalloc(1, sizeof(SFunctParam));

×

1992

    QUERY_CHECK_NULL(pExp->base.pParam, code, lino, _end, terrno);

×

1993

    pExp->base.numOfParams = 1;

×

1994

    SDataType* pType = &pCond->node.resType;

×

1995

    pExp->base.resSchema = createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCond->node.aliasName);

×

1996

    pExp->pExpr->_optrRoot.pRootNode = pNode;

×

1998

    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1999

    QUERY_CHECK_CODE(code, lino, _end);

×

2001

  pExp->pExpr->relatedTo = ((SExprNode*)pNode)->relatedTo;

×

2002

_end:

×

2003

  if (code != TSDB_CODE_SUCCESS) {

×

2004

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

2006

  return code;

×

2009

int32_t createExprFromTargetNode(SExprInfo* pExp, STargetNode* pTargetNode) {

×

2010

  return createExprFromOneNode(pExp, pTargetNode->pExpr, pTargetNode->slotId);

×

2013

SExprInfo* createExpr(SNodeList* pNodeList, int32_t* numOfExprs) {

×

2014

  *numOfExprs = LIST_LENGTH(pNodeList);

×

2015

  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));

×

2016

  if (!pExprs) {

×

2017

    return NULL;

×

2020

  for (int32_t i = 0; i < (*numOfExprs); ++i) {

×

2021

    SExprInfo* pExp = &pExprs[i];

×

2022

    int32_t    code = createExprFromOneNode(pExp, nodesListGetNode(pNodeList, i), i + UD_TAG_COLUMN_INDEX);

×

2023

    if (code != TSDB_CODE_SUCCESS) {

×

2024

      taosMemoryFreeClear(pExprs);

×

2025

      terrno = code;

×

2026

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

2027

      return NULL;

×

2031

  return pExprs;

×

2034

int32_t createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, SExprInfo** pExprInfo, int32_t* numOfExprs) {

×

2035

  QRY_PARAM_CHECK(pExprInfo);

×

2037

  int32_t code = 0;

×

2038

  int32_t numOfFuncs = LIST_LENGTH(pNodeList);

×

2039

  int32_t numOfGroupKeys = 0;

×

2040

  if (pGroupKeys != NULL) {

×

2041

    numOfGroupKeys = LIST_LENGTH(pGroupKeys);

×

2044

  *numOfExprs = numOfFuncs + numOfGroupKeys;

×

2045

  if (*numOfExprs == 0) {

×

2046

    return code;

×

2049

  SExprInfo* pExprs = taosMemoryCalloc(*numOfExprs, sizeof(SExprInfo));

×

2050

  if (pExprs == NULL) {

×

2051

    return terrno;

×

2054

  for (int32_t i = 0; i < (*numOfExprs); ++i) {

×

2055

    STargetNode* pTargetNode = NULL;

×

2056

    if (i < numOfFuncs) {

×

2057

      pTargetNode = (STargetNode*)nodesListGetNode(pNodeList, i);

×

2059

      pTargetNode = (STargetNode*)nodesListGetNode(pGroupKeys, i - numOfFuncs);

×

2061

    if (!pTargetNode) {

×

2062

      destroyExprInfo(pExprs, *numOfExprs);

×

2063

      taosMemoryFreeClear(pExprs);

×

2064

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2065

      return terrno;

×

2068

    SExprInfo* pExp = &pExprs[i];

×

2069

    code = createExprFromTargetNode(pExp, pTargetNode);

×

2070

    if (code != TSDB_CODE_SUCCESS) {

×

2071

      destroyExprInfo(pExprs, *numOfExprs);

×

2072

      taosMemoryFreeClear(pExprs);

×

2073

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

2074

      return code;

×

2078

  *pExprInfo = pExprs;

×

2079

  return code;

×

2082

static void deleteSubsidiareCtx(void* pData) {

×

2083

  SSubsidiaryResInfo* pCtx = (SSubsidiaryResInfo*)pData;

×

2084

  if (pCtx->pCtx) {

×

2085

    taosMemoryFreeClear(pCtx->pCtx);

×

2087

×

2090

static int32_t setSelectValueColumnInfo(SqlFunctionCtx* pCtx, int32_t numOfOutput) {

×

2091

  int32_t num = 0;

×

2092

  int32_t code = TSDB_CODE_SUCCESS;

×

2093

  int32_t lino = 0;

×

2095

  SArray* pValCtxArray = NULL;

×

2096

  for (int32_t i = numOfOutput - 1; i > 0; --i) {  // select Func is at the end of the list

×

2097

    int32_t funcIdx = pCtx[i].pExpr->pExpr->_function.bindExprID;

×

2098

    if (funcIdx > 0) {

×

2099

      if (pValCtxArray == NULL) {

×

2101

        pValCtxArray = taosArrayInit_s(sizeof(SSubsidiaryResInfo*), funcIdx);

×

2102

        if (pValCtxArray == NULL) {

×

2103

          return terrno;

×

2106

      if (funcIdx > pValCtxArray->size) {

×

2107

        qError("funcIdx:%d is out of range", funcIdx);

×

2108

        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);

×

2109

        return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

2111

      SSubsidiaryResInfo* pSubsidiary = &pCtx[i].subsidiaries;

×

2112

      pSubsidiary->pCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);

×

2113

      if (pSubsidiary->pCtx == NULL) {

×

2114

        taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);

×

2115

        return terrno;

×

2117

      pSubsidiary->num = 0;

×

2118

      taosArraySet(pValCtxArray, funcIdx - 1, &pSubsidiary);

×

2122

  SqlFunctionCtx*  p = NULL;

×

2123

  SqlFunctionCtx** pValCtx = NULL;

×

2124

  if (pValCtxArray == NULL) {

×

2125

    pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);

×

2126

    if (pValCtx == NULL) {

×

2127

      QUERY_CHECK_CODE(terrno, lino, _end);

×

2131

  for (int32_t i = 0; i < numOfOutput; ++i) {

×

2132

    const char* pName = pCtx[i].pExpr->pExpr->_function.functionName;

×

2133

    if ((strcmp(pName, "_select_value") == 0)) {

×

2134

      if (pValCtxArray == NULL) {

×

2135

        pValCtx[num++] = &pCtx[i];

×

2137

        int32_t bindFuncIndex = pCtx[i].pExpr->pExpr->relatedTo;  // start from index 1;

×

2138

        if (bindFuncIndex > 0) {  // 0 is default index related to the select function

×

2139

          bindFuncIndex -= 1;

×

2141

        SSubsidiaryResInfo** pSubsidiary = taosArrayGet(pValCtxArray, bindFuncIndex);

×

2142

        if(pSubsidiary == NULL) {

×

2143

          QUERY_CHECK_CODE(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR, lino, _end);

×

2145

        (*pSubsidiary)->pCtx[(*pSubsidiary)->num] = &pCtx[i];

×

2146

        (*pSubsidiary)->num++;

×

2148

    } else if (fmIsSelectFunc(pCtx[i].functionId)) {

×

2149

       if (pValCtxArray == NULL) {

×

2150

        p = &pCtx[i];

×

2155

  if (p != NULL) {

×

2156

    p->subsidiaries.pCtx = pValCtx;

×

2157

    p->subsidiaries.num = num;

×

2159

    taosMemoryFreeClear(pValCtx);

×

2162

_end:

×

2163

  if (code != TSDB_CODE_SUCCESS) {

×

2164

    taosArrayDestroyP(pValCtxArray, deleteSubsidiareCtx);

×

2165

    taosMemoryFreeClear(pValCtx);

×

2166

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

2168

    taosArrayDestroy(pValCtxArray);

×

2170

  return code;

×

2173

SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, int32_t** rowEntryInfoOffset,

×

2175

  int32_t         code = TSDB_CODE_SUCCESS;

×

2176

  int32_t         lino = 0;

×

2177

  SqlFunctionCtx* pFuncCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));

×

2178

  if (pFuncCtx == NULL) {

×

2179

    return NULL;

×

2182

  *rowEntryInfoOffset = taosMemoryCalloc(numOfOutput, sizeof(int32_t));

×

2183

  if (*rowEntryInfoOffset == 0) {

×

2184

    taosMemoryFreeClear(pFuncCtx);

×

2185

    return NULL;

×

2188

  for (int32_t i = 0; i < numOfOutput; ++i) {

×

2189

    SExprInfo* pExpr = &pExprInfo[i];

×

2191

    SExprBasicInfo* pFunct = &pExpr->base;

×

2192

    SqlFunctionCtx* pCtx = &pFuncCtx[i];

×

2194

    pCtx->functionId = -1;

×

2195

    pCtx->pExpr = pExpr;

×

2197

    if (pExpr->pExpr->nodeType == QUERY_NODE_FUNCTION) {

×

2198

      SFuncExecEnv env = {0};

×

2199

      pCtx->functionId = pExpr->pExpr->_function.pFunctNode->funcId;

×

2200

      pCtx->isPseudoFunc = fmIsWindowPseudoColumnFunc(pCtx->functionId);

×

2201

      pCtx->isNotNullFunc = fmIsNotNullOutputFunc(pCtx->functionId);

×

2203

      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);

×

2204

      if (fmIsAggFunc(pCtx->functionId) || fmIsIndefiniteRowsFunc(pCtx->functionId)) {

×

2205

        if (!isUdaf) {

×

2206

          code = fmGetFuncExecFuncs(pCtx->functionId, &pCtx->fpSet);

×

2207

          QUERY_CHECK_CODE(code, lino, _end);

×

2209

          char* udfName = pExpr->pExpr->_function.pFunctNode->functionName;

×

2210

          pCtx->udfName = taosStrdup(udfName);

×

2211

          QUERY_CHECK_NULL(pCtx->udfName, code, lino, _end, terrno);

×

2213

          code = fmGetUdafExecFuncs(pCtx->functionId, &pCtx->fpSet);

×

2214

          QUERY_CHECK_CODE(code, lino, _end);

×

2216

        bool tmp = pCtx->fpSet.getEnv(pExpr->pExpr->_function.pFunctNode, &env);

×

2217

        if (!tmp) {

×

2218

          code = terrno;

×

2219

          QUERY_CHECK_CODE(code, lino, _end);

×

2222

        code = fmGetScalarFuncExecFuncs(pCtx->functionId, &pCtx->sfp);

×

2223

        if (code != TSDB_CODE_SUCCESS && isUdaf) {

×

2224

          code = TSDB_CODE_SUCCESS;

×

2226

        QUERY_CHECK_CODE(code, lino, _end);

×

2228

        if (pCtx->sfp.getEnv != NULL) {

×

2229

          bool tmp = pCtx->sfp.getEnv(pExpr->pExpr->_function.pFunctNode, &env);

×

2230

          if (!tmp) {

×

2231

            code = terrno;

×

2232

            QUERY_CHECK_CODE(code, lino, _end);

×

2236

      pCtx->resDataInfo.interBufSize = env.calcMemSize;

×

2237

    } else if (pExpr->pExpr->nodeType == QUERY_NODE_COLUMN || pExpr->pExpr->nodeType == QUERY_NODE_OPERATOR ||

×

2238

               pExpr->pExpr->nodeType == QUERY_NODE_VALUE) {

×

2240

      pCtx->resDataInfo.interBufSize = pFunct->resSchema.bytes;

×

2243

    pCtx->input.numOfInputCols = pFunct->numOfParams;

×

2244

    pCtx->input.pData = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);

×

2245

    QUERY_CHECK_NULL(pCtx->input.pData, code, lino, _end, terrno);

×

2246

    pCtx->input.pColumnDataAgg = taosMemoryCalloc(pFunct->numOfParams, POINTER_BYTES);

×

2247

    QUERY_CHECK_NULL(pCtx->input.pColumnDataAgg, code, lino, _end, terrno);

×

2249

    pCtx->pTsOutput = NULL;

×

2250

    pCtx->resDataInfo.bytes = pFunct->resSchema.bytes;

×

2251

    pCtx->resDataInfo.type = pFunct->resSchema.type;

×

2252

    pCtx->order = TSDB_ORDER_ASC;

×

2253

    pCtx->start.key = INT64_MIN;

×

2254

    pCtx->end.key = INT64_MIN;

×

2255

    pCtx->numOfParams = pExpr->base.numOfParams;

×

2256

    pCtx->param = pFunct->pParam;

×

2257

    pCtx->saveHandle.currentPage = -1;

×

2258

    pCtx->pStore = pStore;

×

2259

    pCtx->hasWindowOrGroup = false;

×

2260

    pCtx->needCleanup = false;

×

2263

  for (int32_t i = 1; i < numOfOutput; ++i) {

×

2264

    (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) +

×

2265

                                         pFuncCtx[i - 1].resDataInfo.interBufSize);

×

2268

  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);

×

2269

  QUERY_CHECK_CODE(code, lino, _end);

×

2271

_end:

×

2272

  if (code != TSDB_CODE_SUCCESS) {

×

2273

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

2274

    for (int32_t i = 0; i < numOfOutput; ++i) {

×

2275

      taosMemoryFree(pFuncCtx[i].input.pData);

×

2276

      taosMemoryFree(pFuncCtx[i].input.pColumnDataAgg);

×

2278

    taosMemoryFreeClear(*rowEntryInfoOffset);

×

2279

    taosMemoryFreeClear(pFuncCtx);

×

2281

    terrno = code;

×

2282

    return NULL;

×

2284

  return pFuncCtx;

×

2289

int32_t relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray* pCols, bool outputEveryColumn) {

×

2290

  int32_t code = TSDB_CODE_SUCCESS;

×

2291

  size_t  numOfSrcCols = taosArrayGetSize(pCols);

×

2293

  int32_t i = 0, j = 0;

×

2294

  while (i < numOfSrcCols && j < taosArrayGetSize(pColMatchInfo)) {

×

2295

    SColumnInfoData* p = taosArrayGet(pCols, i);

×

2296

    if (!p) {

×

2297

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2298

      return terrno;

×

2300

    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);

×

2301

    if (!pmInfo) {

×

2302

      return terrno;

×

2305

    if (p->info.colId == pmInfo->colId) {

×

2306

      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->dstSlotId);

×

2307

      if (!pDst) {

×

2308

        return terrno;

×

2310

      code = colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);

×

2311

      if (code != TSDB_CODE_SUCCESS) {

×

2312

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

2313

        return code;

×

2315

      i++;

×

2316

      j++;

×

2317

    } else if (p->info.colId < pmInfo->colId) {

×

2318

      i++;

×

2320

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));

×

2321

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

2324

  return code;

×

2327

SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {

×

2328

  SInterval interval = {

×

2329

      .interval = pTableScanNode->interval,

×

2330

      .sliding = pTableScanNode->sliding,

×

2331

      .intervalUnit = pTableScanNode->intervalUnit,

×

2332

      .slidingUnit = pTableScanNode->slidingUnit,

×

2333

      .offset = pTableScanNode->offset,

×

2334

      .precision = pTableScanNode->scan.node.pOutputDataBlockDesc->precision,

×

2337

  calcIntervalAutoOffset(&interval);

×

2339

  return interval;

×

2342

SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {

×

2343

  SColumn c = {0};

×

2345

  c.slotId = pColNode->slotId;

×

2346

  c.colId = pColNode->colId;

×

2347

  c.type = pColNode->node.resType.type;

×

2348

  c.bytes = pColNode->node.resType.bytes;

×

2349

  c.scale = pColNode->node.resType.scale;

×

2350

  c.precision = pColNode->node.resType.precision;

×

2351

  return c;

×

2354

int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode,

×

2356

  pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;

×

2357

  pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);

×

2359

  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));

×

2360

  if (!pCond->colList) {

×

2361

    return terrno;

×

2363

  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);

×

2364

  if (pCond->pSlotList == NULL) {

×

2365

    taosMemoryFreeClear(pCond->colList);

×

2366

    return terrno;

×

2370

  pCond->twindows = pTableScanNode->scanRange;

×

2371

  pCond->suid = pTableScanNode->scan.suid;

×

2372

  pCond->type = TIMEWINDOW_RANGE_CONTAINED;

×

2373

  pCond->startVersion = -1;

×

2374

  pCond->endVersion = -1;

×

2375

  pCond->skipRollup = readHandle->skipRollup;

×

2378

  pCond->notLoadData = (pTableScanNode->dataRequired == FUNC_DATA_REQUIRED_NOT_LOAD) &&

×

2379

                       (pTableScanNode->scan.node.pConditions == NULL) && (pTableScanNode->interval == 0);

×

2381

  int32_t j = 0;

×

2382

  for (int32_t i = 0; i < pCond->numOfCols; ++i) {

×

2383

    STargetNode* pNode = (STargetNode*)nodesListGetNode(pTableScanNode->scan.pScanCols, i);

×

2384

    if (!pNode) {

×

2385

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2386

      return terrno;

×

2388

    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;

×

2389

    if (pColNode->colType == COLUMN_TYPE_TAG) {

×

2390

      continue;

×

2393

    pCond->colList[j].type = pColNode->node.resType.type;

×

2394

    pCond->colList[j].bytes = pColNode->node.resType.bytes;

×

2395

    pCond->colList[j].colId = pColNode->colId;

×

2396

    pCond->colList[j].pk = pColNode->isPk;

×

2398

    pCond->pSlotList[j] = pNode->slotId;

×

2399

    j += 1;

×

2402

  pCond->numOfCols = j;

×

2403

  return TSDB_CODE_SUCCESS;

×

2406

int32_t initQueryTableDataCondWithColArray(SQueryTableDataCond* pCond, SQueryTableDataCond* pOrgCond,

×

2408

  int32_t code = TSDB_CODE_SUCCESS;

×

2409

  int32_t lino = 0;

×

2411

  pCond->order = TSDB_ORDER_ASC;

×

2412

  pCond->numOfCols = (int32_t)taosArrayGetSize(colArray);

×

2414

  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));

×

2415

  QUERY_CHECK_NULL(pCond->colList, code, lino, _return, terrno);

×

2417

  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);

×

2418

  QUERY_CHECK_NULL(pCond->pSlotList, code, lino, _return, terrno);

×

2420

  pCond->twindows = pOrgCond->twindows;

×

2421

  pCond->type = pOrgCond->type;

×

2422

  pCond->startVersion = -1;

×

2423

  pCond->endVersion = -1;

×

2424

  pCond->skipRollup = true;

×

2425

  pCond->notLoadData = false;

×

2427

  for (int32_t i = 0; i < pCond->numOfCols; ++i) {

×

2428

    SColIdPair* pColPair = taosArrayGet(colArray, i);

×

2429

    QUERY_CHECK_NULL(pColPair, code, lino, _return, terrno);

×

2431

    bool find = false;

×

2432

    for (int32_t j = 0; j < pOrgCond->numOfCols; ++j) {

×

2433

      if (pOrgCond->colList[j].colId == pColPair->vtbColId) {

×

2434

        pCond->colList[i].type = pOrgCond->colList[j].type;

×

2435

        pCond->colList[i].bytes = pOrgCond->colList[j].bytes;

×

2436

        pCond->colList[i].colId = pColPair->orgColId;

×

2437

        pCond->colList[i].pk = pOrgCond->colList[j].pk;

×

2438

        pCond->pSlotList[i] = i;

×

2439

        find = true;

×

2440

        break;

×

2443

    QUERY_CHECK_CONDITION(find, code, lino, _return, TSDB_CODE_NOT_FOUND);

×

2446

  return code;

×

2447

_return:

×

2448

  qError("%s failed at line %d since %s", __func__, lino, tstrerror(terrno));

×

2449

  taosMemoryFreeClear(pCond->colList);

×

2450

  taosMemoryFreeClear(pCond->pSlotList);

×

2451

  return code;

×

2454

void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {

×

2455

  taosMemoryFreeClear(pCond->colList);

×

2456

  taosMemoryFreeClear(pCond->pSlotList);

×

2457

×

2459

int32_t convertFillType(int32_t mode) {

×

2460

  int32_t type = TSDB_FILL_NONE;

×

2461

  switch (mode) {

×

2462

    case FILL_MODE_PREV:

×

2463

      type = TSDB_FILL_PREV;

×

2464

      break;

×

2465

    case FILL_MODE_NONE:

×

2466

      type = TSDB_FILL_NONE;

×

2467

      break;

×

2468

    case FILL_MODE_NULL:

×

2469

      type = TSDB_FILL_NULL;

×

2470

      break;

×

2471

    case FILL_MODE_NULL_F:

×

2472

      type = TSDB_FILL_NULL_F;

×

2473

      break;

×

2474

    case FILL_MODE_NEXT:

×

2475

      type = TSDB_FILL_NEXT;

×

2476

      break;

×

2477

    case FILL_MODE_VALUE:

×

2478

      type = TSDB_FILL_SET_VALUE;

×

2479

      break;

×

2480

    case FILL_MODE_VALUE_F:

×

2481

      type = TSDB_FILL_SET_VALUE_F;

×

2482

      break;

×

2483

    case FILL_MODE_LINEAR:

×

2484

      type = TSDB_FILL_LINEAR;

×

2485

      break;

×

2486

    case FILL_MODE_NEAR:

×

2487

      type = TSDB_FILL_NEAR;

×

2488

      break;

×

2489

    default:

×

2490

      type = TSDB_FILL_NONE;

×

2493

  return type;

×

2496

void getInitialStartTimeWindow(SInterval* pInterval, TSKEY ts, STimeWindow* w, bool ascQuery) {

×

2497

  if (ascQuery) {

×

2498

    *w = getAlignQueryTimeWindow(pInterval, ts);

×

2501

    *w = getAlignQueryTimeWindow(pInterval, ts);

×

2503

    int64_t key = w->skey;

×

2504

    while (key < ts) {  // moving towards end

×

2505

      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);

×

2506

      if (key > ts) {

×

2507

        break;

×

2510

      w->skey = key;

×

2512

    w->ekey = taosTimeAdd(w->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;

×

2514

×

2516

static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {

×

2517

  STimeWindow w = {0};

×

2519

  w.skey = taosTimeTruncate(ts, pInterval);

×

2520

  w.ekey = taosTimeGetIntervalEnd(w.skey, pInterval);

×

2521

  return w;

×

2524

STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInterval* pInterval, int32_t order) {

×

2525

  STimeWindow win = *pWindow;

×

2526

  STimeWindow save = win;

×

2527

  while (win.skey <= ts && win.ekey >= ts) {

×

2528

    save = win;

×

2530

    getNextTimeWindow(pInterval, &win, order == TSDB_ORDER_ASC ? TSDB_ORDER_DESC : TSDB_ORDER_ASC);

×

2533

  return save;

×

2538

STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,

×

2540

  STimeWindow w = {0};

×

2541

  if (pResultRowInfo->cur.pageId == -1) {  // the first window, from the previous stored value

×

2542

    getInitialStartTimeWindow(pInterval, ts, &w, (order == TSDB_ORDER_ASC));

×

2543

    return w;

×

2546

  SResultRow* pRow = getResultRowByPos(pBuf, &pResultRowInfo->cur, false);

×

2547

  if (pRow) {

×

2548

    TAOS_SET_OBJ_ALIGNED(&w, pRow->win);

×

2552

  if (w.skey > ts || w.ekey < ts) {

×

2553

    w = doCalculateTimeWindow(ts, pInterval);

×

2556

  if (pInterval->interval != pInterval->sliding) {

×

2559

    w = getFirstQualifiedTimeWindow(ts, &w, pInterval, order);

×

2562

  return w;

×

2565

TSKEY getNextTimeWindowStart(const SInterval* pInterval, TSKEY start, int32_t order) {

×

2566

  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);

×

2567

  TSKEY   nextStart = taosTimeAdd(start, -1 * pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);

×

2568

  nextStart = taosTimeAdd(nextStart, factor * pInterval->sliding, pInterval->slidingUnit, pInterval->precision, NULL);

×

2569

  nextStart = taosTimeAdd(nextStart, pInterval->offset, pInterval->offsetUnit, pInterval->precision, NULL);

×

2570

  return nextStart;

×

2573

void getNextTimeWindow(const SInterval* pInterval, STimeWindow* tw, int32_t order) {

×

2574

  tw->skey = getNextTimeWindowStart(pInterval, tw->skey, order);

×

2575

  tw->ekey = taosTimeAdd(tw->skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision, NULL) - 1;

×

2576

×

2578

bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {

×

2579

  return (pLimitInfo->limit.limit != -1 || pLimitInfo->limit.offset != -1 || pLimitInfo->slimit.limit != -1 ||

×

2580

          pLimitInfo->slimit.offset != -1);

×

2583

bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {

×

2584

  return (pLimitInfo->slimit.limit != -1 || pLimitInfo->slimit.offset != -1);

×

2587

void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo) {

×

2588

  SLimit limit = {.limit = getLimit(pLimit), .offset = getOffset(pLimit)};

×

2589

  SLimit slimit = {.limit = getLimit(pSLimit), .offset = getOffset(pSLimit)};

×

2591

  pLimitInfo->limit = limit;

×

2592

  pLimitInfo->slimit = slimit;

×

2593

  pLimitInfo->remainOffset = limit.offset;

×

2594

  pLimitInfo->remainGroupOffset = slimit.offset;

×

2595

×

2597

void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {

×

2598

  pLimitInfo->numOfOutputRows = 0;

×

2599

  pLimitInfo->remainOffset = pLimitInfo->limit.offset;

×

2600

×

2602

int32_t tableListGetSize(const STableListInfo* pTableList, int32_t* pRes) {

×

2603

  if (taosArrayGetSize(pTableList->pTableList) != taosHashGetSize(pTableList->map)) {

×

2604

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR));

×

2605

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

2607

  (*pRes) = taosArrayGetSize(pTableList->pTableList);

×

2608

  return TSDB_CODE_SUCCESS;

×

2611

uint64_t tableListGetSuid(const STableListInfo* pTableList) { return pTableList->idInfo.suid; }

×

2613

STableKeyInfo* tableListGetInfo(const STableListInfo* pTableList, int32_t index) {

×

2614

  if (taosArrayGetSize(pTableList->pTableList) == 0) {

×

2615

    return NULL;

×

2618

  return taosArrayGet(pTableList->pTableList, index);

×

2621

int32_t tableListFind(const STableListInfo* pTableList, uint64_t uid, int32_t startIndex) {

×

2622

  int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);

×

2623

  if (startIndex >= numOfTables) {

×

2624

    return -1;

×

2627

  for (int32_t i = startIndex; i < numOfTables; ++i) {

×

2628

    STableKeyInfo* p = taosArrayGet(pTableList->pTableList, i);

×

2629

    if (!p) {

×

2630

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2631

      return -1;

×

2633

    if (p->uid == uid) {

×

2634

      return i;

×

2637

  return -1;

×

2640

void tableListGetSourceTableInfo(const STableListInfo* pTableList, uint64_t* psuid, uint64_t* uid, int32_t* type) {

×

2641

  *psuid = pTableList->idInfo.suid;

×

2642

  *uid = pTableList->idInfo.uid;

×

2643

  *type = pTableList->idInfo.tableType;

×

2644

×

2646

uint64_t tableListGetTableGroupId(const STableListInfo* pTableList, uint64_t tableUid) {

×

2647

  int32_t* slot = taosHashGet(pTableList->map, &tableUid, sizeof(tableUid));

×

2648

  if (slot == NULL) {

×

2649

    return -1;

×

2652

  STableKeyInfo* pKeyInfo = taosArrayGet(pTableList->pTableList, *slot);

×

2654

  return pKeyInfo->groupId;

×

2658

int32_t tableListAddTableInfo(STableListInfo* pTableList, uint64_t uid, uint64_t gid) {

×

2659

  int32_t code = TSDB_CODE_SUCCESS;

×

2660

  int32_t lino = 0;

×

2661

  if (pTableList->map == NULL) {

×

2662

    pTableList->map = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

2663

    QUERY_CHECK_NULL(pTableList->map, code, lino, _end, terrno);

×

2666

  STableKeyInfo keyInfo = {.uid = uid, .groupId = gid};

×

2667

  void*         p = taosHashGet(pTableList->map, &uid, sizeof(uid));

×

2668

  if (p != NULL) {

×

2669

    qInfo("table:%" PRId64 " already in tableIdList, ignore it", uid);

×

2670

    goto _end;

×

2673

  void* tmp = taosArrayPush(pTableList->pTableList, &keyInfo);

×

2674

  QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

2676

  int32_t slot = (int32_t)taosArrayGetSize(pTableList->pTableList) - 1;

×

2677

  code = taosHashPut(pTableList->map, &uid, sizeof(uid), &slot, sizeof(slot));

×

2678

  if (code != TSDB_CODE_SUCCESS) {

×

2680

    QUERY_CHECK_CONDITION((code != TSDB_CODE_DUP_KEY), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

2681

    taosArrayPopTailBatch(pTableList->pTableList, 1);  // let's pop the last element in the array list

×

2684

_end:

×

2685

  if (code != TSDB_CODE_SUCCESS) {

×

2686

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

2688

    qDebug("uid:%" PRIu64 ", groupId:%" PRIu64 " added into table list, slot:%d, total:%d", uid, gid, slot, slot + 1);

×

2691

  return code;

×

2694

int32_t tableListGetGroupList(const STableListInfo* pTableList, int32_t ordinalGroupIndex, STableKeyInfo** pKeyInfo,

×

2696

  int32_t totalGroups = tableListGetOutputGroups(pTableList);

×

2697

  int32_t numOfTables = 0;

×

2698

  int32_t code = tableListGetSize(pTableList, &numOfTables);

×

2699

  if (code != TSDB_CODE_SUCCESS) {

×

2700

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

2701

    return code;

×

2704

  if (ordinalGroupIndex < 0 || ordinalGroupIndex >= totalGroups) {

×

2705

    return TSDB_CODE_INVALID_PARA;

×

2710

  if (totalGroups == 1) {

×

2711

    *size = numOfTables;

×

2712

    *pKeyInfo = (*size == 0) ? NULL : taosArrayGet(pTableList->pTableList, 0);

×

2713

    return TSDB_CODE_SUCCESS;

×

2714

  } else if (totalGroups == numOfTables) {

×

2715

    *size = 1;

×

2716

    *pKeyInfo = taosArrayGet(pTableList->pTableList, ordinalGroupIndex);

×

2717

    return TSDB_CODE_SUCCESS;

×

2720

  int32_t offset = pTableList->groupOffset[ordinalGroupIndex];

×

2721

  if (ordinalGroupIndex < totalGroups - 1) {

×

2722

    *size = pTableList->groupOffset[ordinalGroupIndex + 1] - offset;

×

2724

    *size = numOfTables - offset;

×

2727

  *pKeyInfo = taosArrayGet(pTableList->pTableList, offset);

×

2728

  return TSDB_CODE_SUCCESS;

×

2731

int32_t tableListGetOutputGroups(const STableListInfo* pTableList) { return pTableList->numOfOuputGroups; }

×

2733

bool oneTableForEachGroup(const STableListInfo* pTableList) { return pTableList->oneTableForEachGroup; }

×

2735

STableListInfo* tableListCreate() {

×

2736

  STableListInfo* pListInfo = taosMemoryCalloc(1, sizeof(STableListInfo));

×

2737

  if (pListInfo == NULL) {

×

2738

    return NULL;

×

2741

  pListInfo->remainGroups = NULL;

×

2742

  pListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo));

×

2743

  if (pListInfo->pTableList == NULL) {

×

2744

    goto _error;

×

2747

  pListInfo->map = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

2748

  if (pListInfo->map == NULL) {

×

2749

    goto _error;

×

2752

  pListInfo->numOfOuputGroups = 1;

×

2753

  return pListInfo;

×

2755

_error:

×

2756

  tableListDestroy(pListInfo);

×

2757

  return NULL;

×

2760

void tableListDestroy(STableListInfo* pTableListInfo) {

×

2761

  if (pTableListInfo == NULL) {

×

2762

    return;

×

2765

  taosArrayDestroy(pTableListInfo->pTableList);

×

2766

  taosMemoryFreeClear(pTableListInfo->groupOffset);

×

2768

  taosHashCleanup(pTableListInfo->map);

×

2769

  taosHashCleanup(pTableListInfo->remainGroups);

×

2770

  pTableListInfo->pTableList = NULL;

×

2771

  pTableListInfo->map = NULL;

×

2772

  taosMemoryFree(pTableListInfo);

×

2775

void tableListClear(STableListInfo* pTableListInfo) {

×

2776

  if (pTableListInfo == NULL) {

×

2777

    return;

×

2780

  taosArrayClear(pTableListInfo->pTableList);

×

2781

  taosHashClear(pTableListInfo->map);

×

2782

  taosHashClear(pTableListInfo->remainGroups);

×

2783

  taosMemoryFree(pTableListInfo->groupOffset);

×

2784

  pTableListInfo->numOfOuputGroups = 1;

×

2785

  pTableListInfo->oneTableForEachGroup = false;

×

2788

static int32_t orderbyGroupIdComparFn(const void* p1, const void* p2) {

×

2789

  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;

×

2790

  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;

×

2792

  if (pInfo1->groupId == pInfo2->groupId) {

×

2793

    return 0;

×

2795

    return pInfo1->groupId < pInfo2->groupId ? -1 : 1;

×

2799

static int32_t sortTableGroup(STableListInfo* pTableListInfo) {

×

2800

  taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);

×

2801

  int32_t size = taosArrayGetSize(pTableListInfo->pTableList);

×

2803

  SArray* pList = taosArrayInit(4, sizeof(int32_t));

×

2804

  if (!pList) {

×

2805

    return terrno;

×

2808

  STableKeyInfo* pInfo = taosArrayGet(pTableListInfo->pTableList, 0);

×

2809

  if (!pInfo) {

×

2810

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2811

    return terrno;

×

2813

  uint64_t       gid = pInfo->groupId;

×

2815

  int32_t start = 0;

×

2816

  void*   tmp = taosArrayPush(pList, &start);

×

2817

  if (!tmp) {

×

2818

    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2819

    return terrno;

×

2822

  for (int32_t i = 1; i < size; ++i) {

×

2823

    pInfo = taosArrayGet(pTableListInfo->pTableList, i);

×

2824

    if (!pInfo) {

×

2825

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2826

      return terrno;

×

2828

    if (pInfo->groupId != gid) {

×

2829

      tmp = taosArrayPush(pList, &i);

×

2830

      if (!tmp) {

×

2831

        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2832

        return terrno;

×

2834

      gid = pInfo->groupId;

×

2838

  pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);

×

2839

  pTableListInfo->groupOffset = taosMemoryMalloc(sizeof(int32_t) * pTableListInfo->numOfOuputGroups);

×

2840

  if (pTableListInfo->groupOffset == NULL) {

×

2841

    taosArrayDestroy(pList);

×

2842

    return terrno;

×

2845

  memcpy(pTableListInfo->groupOffset, taosArrayGet(pList, 0), sizeof(int32_t) * pTableListInfo->numOfOuputGroups);

×

2846

  taosArrayDestroy(pList);

×

2847

  return TSDB_CODE_SUCCESS;

×

2850

int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle* pHandle, SScanPhysiNode* pScanNode,

×

2852

  int32_t code = TSDB_CODE_SUCCESS;

×

2854

  bool   groupByTbname = groupbyTbname(group);

×

2855

  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);

×

2856

  if (!numOfTables) {

×

2857

    return code;

×

2859

  if (group == NULL || groupByTbname) {

×

2860

    if (tsCountAlwaysReturnValue && QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode) &&

×

2861

        ((STableScanPhysiNode*)pScanNode)->needCountEmptyTable) {

×

2862

      pTableListInfo->remainGroups =

×

2863

          taosHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

×

2864

      if (pTableListInfo->remainGroups == NULL) {

×

2865

        return terrno;

×

2868

      for (int i = 0; i < numOfTables; i++) {

×

2869

        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);

×

2870

        if (!info) {

×

2871

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2872

          return terrno;

×

2874

        info->groupId = groupByTbname ? info->uid : 0;

×

2876

        int32_t tempRes = taosHashPut(pTableListInfo->remainGroups, &(info->groupId), sizeof(info->groupId),

×

2877

                                      &(info->uid), sizeof(info->uid));

×

2878

        if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {

×

2879

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));

×

2880

          return tempRes;

×

2884

      for (int32_t i = 0; i < numOfTables; i++) {

×

2885

        STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);

×

2886

        if (!info) {

×

2887

          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2888

          return terrno;

×

2890

        info->groupId = groupByTbname ? info->uid : 0;

×

2894

    pTableListInfo->oneTableForEachGroup = groupByTbname;

×

2895

    if (numOfTables == 1 && pTableListInfo->idInfo.tableType == TSDB_CHILD_TABLE) {

×

2896

      pTableListInfo->oneTableForEachGroup = true;

×

2899

    if (groupSort && groupByTbname) {

×

2900

      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);

×

2901

      pTableListInfo->numOfOuputGroups = numOfTables;

×

2902

    } else if (groupByTbname && pScanNode->groupOrderScan) {

×

2903

      pTableListInfo->numOfOuputGroups = numOfTables;

×

2905

      pTableListInfo->numOfOuputGroups = 1;

×

2908

    bool initRemainGroups = false;

×

2909

    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {

×

2910

      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;

×

2911

      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&

×

2912

          !(groupSort || pScanNode->groupOrderScan)) {

×

2913

        initRemainGroups = true;

×

2917

    code = getColInfoResultForGroupby(pHandle->vnode, group, pTableListInfo, digest, pAPI, initRemainGroups);

×

2918

    if (code != TSDB_CODE_SUCCESS) {

×

2919

      return code;

×

2922

    if (pScanNode->groupOrderScan) pTableListInfo->numOfOuputGroups = taosArrayGetSize(pTableListInfo->pTableList);

×

2924

    if (groupSort || pScanNode->groupOrderScan) {

×

2925

      code = sortTableGroup(pTableListInfo);

×

2930

  size_t size = taosArrayGetSize(pTableListInfo->pTableList);

×

2931

  for (int32_t i = 0; i < size; ++i) {

×

2932

    STableKeyInfo* p = taosArrayGet(pTableListInfo->pTableList, i);

×

2933

    if (!p) {

×

2934

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));

×

2935

      return terrno;

×

2937

    int32_t        tempRes = taosHashPut(pTableListInfo->map, &p->uid, sizeof(uint64_t), &i, sizeof(int32_t));

×

2938

    if (tempRes != TSDB_CODE_SUCCESS && tempRes != TSDB_CODE_DUP_KEY) {

×

2939

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(tempRes));

×

2940

      return tempRes;

×

2944

  return code;

×

2947

int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,

×

2950

  int64_t     st = taosGetTimestampUs();

×

2951

  const char* idStr = GET_TASKID(pTaskInfo);

×

2953

  if (pHandle == NULL) {

×

2954

    qError("invalid handle, in creating operator tree, %s", idStr);

×

2955

    return TSDB_CODE_INVALID_PARA;

×

2958

  uint8_t digest[17] = {0};

×

2959

  int32_t code = getTableList(pHandle->vnode, pScanNode, pTagCond, pTagIndexCond, pTableListInfo, digest, idStr,

×

2961

  if (code != TSDB_CODE_SUCCESS) {

×

2962

    qError("failed to getTableList, code:%s", tstrerror(code));

×

2963

    return code;

×

2966

  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);

×

2968

  int64_t st1 = taosGetTimestampUs();

×

2969

  pTaskInfo->cost.extractListTime = (st1 - st) / 1000.0;

×

2970

  qDebug("extract queried table list completed, %d tables, elapsed time:%.2f ms %s", numOfTables,

×

2973

  if (numOfTables == 0) {

×

2974

    qDebug("no table qualified for query, %s" PRIx64, idStr);

×

2975

    return TSDB_CODE_SUCCESS;

×

2978

  code = buildGroupIdMapForAllTables(pTableListInfo, pHandle, pScanNode, pGroupTags, groupSort, digest,

×

2980

  if (code != TSDB_CODE_SUCCESS) {

×

2981

    return code;

×

2984

  pTaskInfo->cost.groupIdMapTime = (taosGetTimestampUs() - st1) / 1000.0;

×

2985

  qDebug("generate group id map completed, elapsed time:%.2f ms %s", pTaskInfo->cost.groupIdMapTime, idStr);

×

2987

  return TSDB_CODE_SUCCESS;

×

2990

char* getStreamOpName(uint16_t opType) {

×

2991

  switch (opType) {

×

2992

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:

×

2993

      return "stream scan";

×

2994

    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:

×

2995

      return "project";

×

2996

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:

×

2997

      return "interval single";

×

2998

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:

×

2999

      return "interval final";

×

3000

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:

×

3001

      return "interval semi";

×

3002

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL:

×

3003

      return "interval mid";

×

3004

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:

×

3005

      return "stream fill";

×

3006

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:

×

3007

      return "session single";

×

3008

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION:

×

3009

      return "session semi";

×

3010

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION:

×

3011

      return "session final";

×

3012

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:

×

3013

      return "state single";

×

3014

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:

×

3015

      return "stream partitionby";

×

3016

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT:

×

3017

      return "stream event";

×

3018

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT:

×

3019

      return "stream count";

×

3020

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC:

×

3021

      return "stream interp";

×

3022

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL:

×

3023

      return "interval continue";

×

3024

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_INTERVAL:

×

3025

      return "interval continue semi";

×

3026

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_INTERVAL:

×

3027

      return "interval continue final";

×

3028

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SESSION:

×

3029

      return "session continue";

×

3030

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_SEMI_SESSION:

×

3031

      return "session continue semi";

×

3032

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_FINAL_SESSION:

×

3033

      return "session continue final";

×

3034

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_STATE:

×

3035

      return "state continue";

×

3036

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_EVENT:

×

3037

      return "event continue";

×

3038

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_COUNT:

×

3039

      return "count continue";

×

3041

  return "error name";

×

3044

void printDataBlock(SSDataBlock* pBlock, const char* flag, const char* taskIdStr) {

×

3045

  if (!pBlock) {

×

3046

    qInfo("%s===stream===%s: Block is Null", taskIdStr, flag);

×

3047

    return;

×

3048

  } else if (pBlock->info.rows == 0) {

×

3049

    qInfo("%s===stream===%s: Block is Empty. block type %d", taskIdStr, flag, pBlock->info.type);

×

3050

    return;

×

3052

  if (qDebugFlag & DEBUG_INFO) {

×

3053

    char*   pBuf = NULL;

×

3054

    int32_t code = dumpBlockData(pBlock, flag, &pBuf, taskIdStr);

×

3055

    if (code == 0) {

×

3056

      qInfo("%s", pBuf);

×

3057

      taosMemoryFree(pBuf);

×

3062

void printSpecDataBlock(SSDataBlock* pBlock, const char* flag, const char* opStr, const char* taskIdStr) {

×

3063

  if (!pBlock) {

×

3064

    qDebug("%s===stream===%s %s: Block is Null", taskIdStr, flag, opStr);

×

3065

    return;

×

3066

  } else if (pBlock->info.rows == 0) {

×

3067

    qDebug("%s===stream===%s %s: Block is Empty. block type %d.skey:%" PRId64 ",ekey:%" PRId64 ",version%" PRId64,

×

3070

    return;

×

3072

  if (qDebugFlag & DEBUG_INFO) {

×

3073

    char* pBuf = NULL;

×

3075

    snprintf(flagBuf, sizeof(flagBuf), "%s %s", flag, opStr);

×

3076

    int32_t code = dumpBlockData(pBlock, flagBuf, &pBuf, taskIdStr);

×

3077

    if (code == 0) {

×

3078

      qInfo("%s", pBuf);

×

3079

      taosMemoryFree(pBuf);

×

3084

TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }

×

3086

void updateTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pWin, int64_t delta) {

×

3087

  int64_t* ts = (int64_t*)pColData->pData;

×

3089

  int64_t duration = pWin->ekey - pWin->skey + delta;

×

3090

  ts[2] = duration;            // set the duration

×

3091

  ts[3] = pWin->skey;          // window start key

×

3092

  ts[4] = pWin->ekey + delta;  // window end key

×

3093

×

3095

int32_t compKeys(const SArray* pSortGroupCols, const char* oldkeyBuf, int32_t oldKeysLen, const SSDataBlock* pBlock,

×

3097

  SColumnDataAgg* pColAgg = NULL;

×

3098

  const char*     isNull = oldkeyBuf;

×

3099

  const char*     p = oldkeyBuf + sizeof(int8_t) * pSortGroupCols->size;

×

3101

  for (int32_t i = 0; i < pSortGroupCols->size; ++i) {

×

3102

    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);

×

3103

    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);

×

3104

    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];

×

3106

    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {

×

3107

      if (isNull[i] != 1) return 1;

×

3109

      if (isNull[i] != 0) return 1;

×

3110

      const char* val = colDataGetData(pColInfoData, rowIndex);

×

3111

      if (pCol->type == TSDB_DATA_TYPE_JSON) {

×

3112

        int32_t len = getJsonValueLen(val);

×

3113

        if (memcmp(p, val, len) != 0) return 1;

×

3114

        p += len;

×

3115

      } else if (IS_VAR_DATA_TYPE(pCol->type)) {

×

3116

        if (memcmp(p, val, varDataTLen(val)) != 0) return 1;

×

3117

        p += varDataTLen(val);

×

3119

        if (0 != memcmp(p, val, pCol->bytes)) return 1;

×

3120

        p += pCol->bytes;

×

3124

  if ((int32_t)(p - oldkeyBuf) != oldKeysLen) return 1;

×

3125

  return 0;

×

3128

int32_t buildKeys(char* keyBuf, const SArray* pSortGroupCols, const SSDataBlock* pBlock, int32_t rowIndex) {

×

3129

  uint32_t        colNum = pSortGroupCols->size;

×

3130

  SColumnDataAgg* pColAgg = NULL;

×

3131

  char*           isNull = keyBuf;

×

3132

  char*           p = keyBuf + sizeof(int8_t) * colNum;

×

3134

  for (int32_t i = 0; i < colNum; ++i) {

×

3135

    const SColumn*         pCol = (SColumn*)TARRAY_GET_ELEM(pSortGroupCols, i);

×

3136

    const SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, pCol->slotId);

×

3137

    if (pCol->slotId > pBlock->pDataBlock->size) continue;

×

3139

    if (pBlock->pBlockAgg) pColAgg = &pBlock->pBlockAgg[pCol->slotId];

×

3141

    if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {

×

3142

      isNull[i] = 1;

×

3144

      isNull[i] = 0;

×

3145

      const char* val = colDataGetData(pColInfoData, rowIndex);

×

3146

      if (pCol->type == TSDB_DATA_TYPE_JSON) {

×

3147

        int32_t len = getJsonValueLen(val);

×

3148

        memcpy(p, val, len);

×

3149

        p += len;

×

3150

      } else if (IS_VAR_DATA_TYPE(pCol->type)) {

×

3151

        varDataCopy(p, val);

×

3152

        p += varDataTLen(val);

×

3154

        memcpy(p, val, pCol->bytes);

×

3155

        p += pCol->bytes;

×

3159

  return (int32_t)(p - keyBuf);

×

3162

uint64_t calcGroupId(char* pData, int32_t len) {

×

3164

  tMD5Init(&context);

×

3165

  tMD5Update(&context, (uint8_t*)pData, len);

×

3166

  tMD5Final(&context);

×

3169

  uint64_t id = 0;

×

3170

  memcpy(&id, context.digest, sizeof(uint64_t));

×

3171

  if (0 == id) memcpy(&id, context.digest + 8, sizeof(uint64_t));

×

3172

  return id;

×

3175

SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {

×

3177

  SNodeList* ret = NULL;

×

3178

  FOREACH(node, pSortKeys) {

×

3179

    SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;

×

3180

    int32_t           code = nodesListMakeAppend(&ret, pSortKey->pExpr);

×

3181

    if (code != TSDB_CODE_SUCCESS) {

×

3182

      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));

×

3183

      terrno = code;

×

3184

      return NULL;

×

3187

  return ret;

×

3190

int32_t extractKeysLen(const SArray* keys, int32_t* pLen) {

×

3191

  int32_t code = TSDB_CODE_SUCCESS;

×

3192

  int32_t lino = 0;

×

3193

  int32_t len = 0;

×

3194

  int32_t keyNum = taosArrayGetSize(keys);

×

3195

  for (int32_t i = 0; i < keyNum; ++i) {

×

3196

    SColumn* pCol = (SColumn*)taosArrayGet(keys, i);

×

3197

    QUERY_CHECK_NULL(pCol, code, lino, _end, terrno);

×

3198

    len += pCol->bytes;

×

3200

  len += sizeof(int8_t) * keyNum;  // null flag

×

3201

  *pLen = len;

×

3203

_end:

×

3204

  if (code != TSDB_CODE_SUCCESS) {

×

3205

    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));

×

3207

  return code;

×

taosdata / TDengine / #3798

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

taosdata / TDengine / #3798

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

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