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

57

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

×

58

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

×

61

void initResultRowInfo(SResultRowInfo* pResultRowInfo) {

×

62

  pResultRowInfo->size = 0;

×

63

  pResultRowInfo->cur.pageId = -1;

×

64

×

66

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

×

68

void resetResultRow(SResultRow* pResultRow, size_t entrySize) {

×

69

  pResultRow->numOfRows = 0;

×

70

  pResultRow->closed = false;

×

71

  pResultRow->endInterp = false;

×

72

  pResultRow->startInterp = false;

×

74

  if (entrySize > 0) {

×

75

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

×

77

×

80

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

×

81

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

×

84

size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {

×

85

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

×

87

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

×

88

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

×

91

  return rowSize;

×

95

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

×

96

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

×

97

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

×

98

    return TSDB_CODE_INVALID_PARA;

×

100

  SqlFunctionCtx *pCtx = pSup->pCtx;

×

101

  int32_t        *offset = pSup->rowEntryInfoOffset;

×

102

  SResultRow     *pResultRow  = NULL;

×

103

  size_t          processedSize = 0;

×

104

  int32_t         code = TSDB_CODE_SUCCESS;

×

107

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

×

108

  *outBuf = taosMemoryMalloc(*outBufSize);

×

109

  if (*outBuf == NULL) {

×

110

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

×

111

    return terrno;

×

113

  pResultRow = (SResultRow*)*outBuf;

×

114

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

×

115

  inBuf += sizeof(SResultRow);

×

116

  processedSize += sizeof(SResultRow);

×

118

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

×

119

    int32_t len = *(int32_t*)inBuf;

×

120

    inBuf += sizeof(int32_t);

×

121

    processedSize += sizeof(int32_t);

×

122

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

×

123

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

×

124

      if (code != TSDB_CODE_SUCCESS) {

×

125

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

×

126

        return code;

×

129

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

×

131

    inBuf += len;

×

132

    processedSize += len;

×

135

  if (processedSize < inBufSize) {

×

137

    size_t leftLen = inBufSize - processedSize;

×

138

    TAOS_MEMORY_REALLOC(*outBuf, *outBufSize + leftLen);

×

139

    if (*outBuf == NULL) {

×

140

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

×

141

      return terrno;

×

143

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

×

144

    inBuf += leftLen;

×

145

    processedSize += leftLen;

×

146

    *outBufSize += leftLen;

×

148

  return TSDB_CODE_SUCCESS;

×

152

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

×

153

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

×

154

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

×

155

    return TSDB_CODE_INVALID_PARA;

×

158

  SqlFunctionCtx *pCtx = pSup->pCtx;

×

159

  int32_t        *offset = pSup->rowEntryInfoOffset;

×

160

  SResultRow     *pResultRow = (SResultRow*)inBuf;

×

161

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

×

163

  if (rowSize > inBufSize) {

×

164

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

×

165

    return TSDB_CODE_INVALID_PARA;

×

169

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

×

170

  if (rowSize < inBufSize) {

×

171

    *outBufSize += inBufSize - rowSize;

×

174

  *outBuf = taosMemoryMalloc(*outBufSize);

×

175

  if (*outBuf == NULL) {

×

176

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

×

177

    return terrno;

×

180

  char *pBuf = *outBuf;

×

181

  pResultRow->version = FUNCTION_RESULT_INFO_VERSION;

×

182

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

×

183

  pBuf += sizeof(SResultRow);

×

184

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

×

185

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

×

186

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

×

187

    pBuf += sizeof(int32_t);

×

188

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

×

189

    pBuf += len;

×

192

  if (rowSize < inBufSize) {

×

194

    size_t leftLen = inBufSize - rowSize;

×

195

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

×

196

    pBuf += leftLen;

×

198

  return TSDB_CODE_SUCCESS;

×

201

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

×

203

void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {

×

204

  taosMemoryFreeClear(pGroupResInfo->pBuf);

×

205

  if (pGroupResInfo->freeItem) {

×

207

    taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);

×

208

    pGroupResInfo->freeItem = false;

×

209

    pGroupResInfo->pRows = NULL;

×

211

    taosArrayDestroy(pGroupResInfo->pRows);

×

212

    pGroupResInfo->pRows = NULL;

×

214

  pGroupResInfo->index = 0;

×

215

×

217

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

×

218

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

×

219

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

×

221

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

×

222

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

×

223

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

×

225

    if (pts1 == pts2) {

×

226

      return 0;

×

228

      return pts1 < pts2 ? -1 : 1;

×

231

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

×

235

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

×

237

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

×

238

  int32_t code = TSDB_CODE_SUCCESS;

×

239

  int32_t lino = 0;

×

240

  if (pGroupResInfo->pRows != NULL) {

×

241

    taosArrayDestroy(pGroupResInfo->pRows);

×

243

  if (pGroupResInfo->pBuf) {

×

244

    taosMemoryFree(pGroupResInfo->pBuf);

×

245

    pGroupResInfo->pBuf = NULL;

×

249

  int32_t size = tSimpleHashGetSize(pHashmap);

×

251

  void* pData = NULL;

×

252

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

×

253

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

×

255

  size_t  keyLen = 0;

×

256

  int32_t iter = 0;

×

257

  int64_t bufLen = 0, offset = 0;

×

260

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

×

262

    bufLen += keyLen + sizeof(SResultRowPosition);

×

265

  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);

×

266

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

×

268

  iter = 0;

×

269

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

×

270

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

×

272

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

×

274

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

×

275

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

×

276

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

×

277

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

×

278

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

×

280

    offset += keyLen + sizeof(struct SResultRowPosition);

×

283

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

×

284

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

×

285

    size = POINTER_BYTES;

×

286

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

×

289

  pGroupResInfo->index = 0;

×

291

_end:

×

292

  if (code != TSDB_CODE_SUCCESS) {

×

293

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

×

295

  return code;

×

298

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

×

299

  if (pGroupResInfo->pRows != NULL) {

×

300

    taosArrayDestroy(pGroupResInfo->pRows);

×

303

  pGroupResInfo->freeItem = true;

×

304

  pGroupResInfo->pRows = pArrayList;

×

305

  pGroupResInfo->index = 0;

×

306

×

308

bool hasRemainResults(SGroupResInfo* pGroupResInfo) {

×

309

  if (pGroupResInfo->pRows == NULL) {

×

310

    return false;

×

313

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

×

316

int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {

×

317

  if (pGroupResInfo->pRows == 0) {

×

318

    return 0;

×

321

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

×

324

SArray* createSortInfo(SNodeList* pNodeList) {

×

325

  size_t numOfCols = 0;

×

327

  if (pNodeList != NULL) {

×

328

    numOfCols = LIST_LENGTH(pNodeList);

×

330

    numOfCols = 0;

×

333

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

×

334

  if (pList == NULL) {

×

335

    return pList;

×

338

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

×

339

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

×

340

    if (!pSortKey) {

×

341

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

×

342

      taosArrayDestroy(pList);

×

343

      pList = NULL;

×

344

      terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

345

      break;

×

347

    SBlockOrderInfo   bi = {0};

×

348

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

×

349

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

×

351

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

×

352

    bi.slotId = pColNode->slotId;

×

353

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

×

354

    if (!tmp) {

×

355

      taosArrayDestroy(pList);

×

356

      pList = NULL;

×

357

      break;

×

361

  return pList;

×

364

SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {

×

365

  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);

×

366

  SSDataBlock* pBlock = NULL;

×

367

  int32_t      code = createDataBlock(&pBlock);

×

368

  if (code) {

×

369

    terrno = code;

×

370

    return NULL;

×

373

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

×

374

  pBlock->info.type = STREAM_INVALID;

×

375

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

×

376

  pBlock->info.watermark = INT64_MIN;

×

378

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

×

379

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

×

380

    if (!pDescNode) {

×

381

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

×

382

      blockDataDestroy(pBlock);

×

383

      pBlock = NULL;

×

384

      terrno = TSDB_CODE_INVALID_PARA;

×

385

      break;

×

388

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

×

389

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

×

390

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

×

391

    idata.info.noData = pDescNode->reserve;

×

393

    code = blockDataAppendColInfo(pBlock, &idata);

×

394

    if (code != TSDB_CODE_SUCCESS) {

×

395

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

×

396

      blockDataDestroy(pBlock);

×

397

      pBlock = NULL;

×

398

      terrno = code;

×

399

      break;

×

403

  return pBlock;

×

406

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

×

407

  SDataBlockInfo* pBlockInfo = &pDataBlock->info;

×

409

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

×

410

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

×

411

    if (!pItem) {

×

412

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

×

413

      return terrno;

×

416

    if (pItem->isPk) {

×

417

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

×

418

      if (!pInfoData) {

×

419

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

×

420

        return terrno;

×

422

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

×

423

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

×

426

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

×

427

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

×

428

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

×

429

          return terrno;

×

432

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

×

433

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

×

434

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

×

435

          return terrno;

×

438

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

×

439

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

×

442

      break;

×

446

  return TSDB_CODE_SUCCESS;

×

449

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

×

450

  STransTagExprCtx* pCtx = pContext;

×

451

  SMetaReader*      mr = pCtx->pReader;

×

452

  bool              isTagCol = false, isTbname = false;

×

453

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

×

454

    SColumnNode* pCol = (SColumnNode*)*pNode;

×

455

    if (pCol->colType == COLUMN_TYPE_TBNAME)

×

456

      isTbname = true;

×

458

      isTagCol = true;

×

459

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

×

460

    SFunctionNode* pFunc = (SFunctionNode*)*pNode;

×

461

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

×

463

  if (isTagCol) {

×

464

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

×

466

    SValueNode* res = NULL;

×

467

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

×

468

    if (NULL == res) {

×

469

      return DEAL_RES_ERROR;

×

472

    res->translate = true;

×

473

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

×

475

    STagVal tagVal = {0};

×

476

    tagVal.cid = pSColumnNode->colId;

×

477

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

×

478

    if (p == NULL) {

×

479

      res->node.resType.type = TSDB_DATA_TYPE_NULL;

×

480

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

×

481

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

×

482

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

×

483

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

×

484

        return DEAL_RES_ERROR;

×

486

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

×

487

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

×

488

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

×

489

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

×

490

        return DEAL_RES_ERROR;

×

492

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

×

493

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

×

495

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

×

496

      if (code != TSDB_CODE_SUCCESS) {

×

497

        return DEAL_RES_ERROR;

×

500

    nodesDestroyNode(*pNode);

×

501

    *pNode = (SNode*)res;

×

502

  } else if (isTbname) {

×

503

    SValueNode* res = NULL;

×

504

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

×

505

    if (NULL == res) {

×

506

      return DEAL_RES_ERROR;

×

509

    res->translate = true;

×

510

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

×

512

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

×

513

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

×

514

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

×

515

      return DEAL_RES_ERROR;

×

517

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

×

518

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

×

519

    nodesDestroyNode(*pNode);

×

520

    *pNode = (SNode*)res;

×

523

  return DEAL_RES_CONTINUE;

×

526

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

×

527

  int32_t     code = TSDB_CODE_SUCCESS;

×

528

  SMetaReader mr = {0};

×

530

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

×

531

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

×

532

  if (TSDB_CODE_SUCCESS != code) {

×

533

    pAPI->metaReaderFn.clearReader(&mr);

×

534

    *pQualified = false;

×

536

    return TSDB_CODE_SUCCESS;

×

539

  SNode* pTagCondTmp = NULL;

×

540

  code = nodesCloneNode(pTagCond, &pTagCondTmp);

×

541

  if (TSDB_CODE_SUCCESS != code) {

×

542

    *pQualified = false;

×

543

    pAPI->metaReaderFn.clearReader(&mr);

×

544

    return code;

×

546

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

×

547

  nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);

×

548

  pAPI->metaReaderFn.clearReader(&mr);

×

549

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

550

    *pQualified = false;

×

551

    terrno = code;

×

552

    return code;

×

555

  SNode* pNew = NULL;

×

556

  code = scalarCalculateConstants(pTagCondTmp, &pNew);

×

557

  if (TSDB_CODE_SUCCESS != code) {

×

558

    terrno = code;

×

559

    nodesDestroyNode(pTagCondTmp);

×

560

    *pQualified = false;

×

562

    return code;

×

565

  SValueNode* pValue = (SValueNode*)pNew;

×

566

  *pQualified = pValue->datum.b;

×

568

  nodesDestroyNode(pNew);

×

569

  return TSDB_CODE_SUCCESS;

×

572

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

×

573

  tagFilterAssist* pData = (tagFilterAssist*)pContext;

×

574

  SColumnNode*     pSColumnNode = NULL;

×

575

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

×

576

    pSColumnNode = *(SColumnNode**)pNode;

×

577

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

×

578

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

×

579

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

×

580

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

×

581

      if (NULL == pSColumnNode) {

×

582

        return DEAL_RES_ERROR;

×

584

      pSColumnNode->colId = -1;

×

585

      pSColumnNode->colType = COLUMN_TYPE_TBNAME;

×

586

      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;

×

587

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

×

588

      nodesDestroyNode(*pNode);

×

589

      *pNode = (SNode*)pSColumnNode;

×

591

      return DEAL_RES_CONTINUE;

×

594

    return DEAL_RES_CONTINUE;

×

597

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

×

598

  if (!data) {

×

600

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

×

601

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

×

602

      return DEAL_RES_ERROR;

×

604

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

×

605

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

×

606

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

×

607

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

×

608

                         .pk = pSColumnNode->isPk};

×

612

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

×

613

    if (!tmp) {

×

614

      return DEAL_RES_ERROR;

×

617

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

×

618

    pSColumnNode->slotId = col->slotId;

×

621

  return DEAL_RES_CONTINUE;

×

624

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

×

625

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

×

626

  if (pColumnData == NULL) {

×

627

    return terrno;

×

630

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

×

631

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

×

632

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

×

633

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

×

635

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

×

636

  if (code != TSDB_CODE_SUCCESS) {

×

637

    terrno = code;

×

638

    releaseColInfoData(pColumnData);

×

639

    return terrno;

×

642

  pParam->columnData = pColumnData;

×

643

  pParam->colAlloced = true;

×

644

  return TSDB_CODE_SUCCESS;

×

647

static void releaseColInfoData(void* pCol) {

×

648

  if (pCol) {

×

649

    SColumnInfoData* col = (SColumnInfoData*)pCol;

×

650

    colDataDestroy(col);

×

651

    taosMemoryFree(col);

×

653

×

655

void freeItem(void* p) {

×

656

  STUidTagInfo* pInfo = p;

×

657

  if (pInfo->pTagVal != NULL) {

×

658

    taosMemoryFree(pInfo->pTagVal);

×

660

×

662

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

×

663

  if (pTagCond == NULL) {

×

664

    return TSDB_CODE_SUCCESS;

×

667

  char*   payload = NULL;

×

668

  int32_t len = 0;

×

669

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

×

670

  if (code != TSDB_CODE_SUCCESS) {

×

671

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

×

672

    return code;

×

675

  tMD5Init(pContext);

×

676

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

×

677

  tMD5Final(pContext);

×

679

  taosMemoryFree(payload);

×

680

  return TSDB_CODE_SUCCESS;

×

683

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

×

684

  int32_t code = TSDB_CODE_SUCCESS;

×

685

  int32_t lino = 0;

×

686

  char*   payload = NULL;

×

687

  int32_t len = 0;

×

688

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

×

689

  QUERY_CHECK_CODE(code, lino, _end);

×

691

  if (filterDigest[0]) {

×

692

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

×

693

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

×

694

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

×

695

    len += tListLen(pContext->digest);

×

698

  tMD5Init(pContext);

×

699

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

×

700

  tMD5Final(pContext);

×

702

_end:

×

703

  taosMemoryFree(payload);

×

704

  if (code != TSDB_CODE_SUCCESS) {

×

705

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

×

707

  return code;

×

710

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

×

712

  int32_t      code = TSDB_CODE_SUCCESS;

×

713

  int32_t      lino = 0;

×

714

  SArray*      pBlockList = NULL;

×

715

  SSDataBlock* pResBlock = NULL;

×

716

  void*        keyBuf = NULL;

×

717

  SArray*      groupData = NULL;

×

718

  SArray*      pUidTagList = NULL;

×

719

  SArray*      tableList = NULL;

×

721

  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);

×

722

  if (rows == 0) {

×

723

    return TSDB_CODE_SUCCESS;

×

726

  tagFilterAssist ctx = {0};

×

727

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

×

728

  if (ctx.colHash == NULL) {

×

729

    code = terrno;

×

730

    goto end;

×

733

  ctx.index = 0;

×

734

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

×

735

  if (ctx.cInfoList == NULL) {

×

736

    code = terrno;

×

737

    goto end;

×

740

  SNode* pNode = NULL;

×

741

  FOREACH(pNode, group) {

×

742

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

×

743

    if (TSDB_CODE_SUCCESS != ctx.code) {

×

744

      code = ctx.code;

×

745

      goto end;

×

747

    REPLACE_NODE(pNode);

×

750

  T_MD5_CTX context = {0};

×

751

  if (tsTagFilterCache) {

×

752

    SNodeListNode* listNode = NULL;

×

753

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

×

754

    if (TSDB_CODE_SUCCESS != code) {

×

755

      goto end;

×

757

    listNode->pNodeList = group;

×

758

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

×

759

    QUERY_CHECK_CODE(code, lino, end);

×

761

    nodesFree(listNode);

×

763

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

×

765

    QUERY_CHECK_CODE(code, lino, end);

×

767

    if (tableList) {

×

768

      taosArrayDestroy(pTableListInfo->pTableList);

×

769

      pTableListInfo->pTableList = tableList;

×

770

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

×

772

      goto end;

×

776

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

×

777

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

×

779

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

×

780

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

×

781

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

×

782

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

×

783

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

×

784

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

×

787

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

×

788

  if (code != TSDB_CODE_SUCCESS) {

×

789

    goto end;

×

792

  int32_t numOfTables = taosArrayGetSize(pUidTagList);

×

793

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

×

794

  if (pResBlock == NULL) {

×

795

    code = terrno;

×

796

    goto end;

×

802

  pBlockList = taosArrayInit(2, POINTER_BYTES);

×

803

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

×

805

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

×

806

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

×

808

  groupData = taosArrayInit(2, POINTER_BYTES);

×

809

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

×

811

  FOREACH(pNode, group) {

×

812

    SScalarParam output = {0};

×

814

    switch (nodeType(pNode)) {

×

815

      case QUERY_NODE_VALUE:

×

816

        break;

×

817

      case QUERY_NODE_COLUMN:

×

820

        SExprNode* expNode = (SExprNode*)pNode;

×

821

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

×

822

        if (code != TSDB_CODE_SUCCESS) {

×

823

          goto end;

×

825

        break;

×

828

      default:

×

829

        code = TSDB_CODE_OPS_NOT_SUPPORT;

×

830

        goto end;

×

833

    if (nodeType(pNode) == QUERY_NODE_COLUMN) {

×

834

      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;

×

835

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

×

836

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

×

837

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

×

838

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

×

839

      continue;

×

841

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

×

844

    if (code != TSDB_CODE_SUCCESS) {

×

845

      releaseColInfoData(output.columnData);

×

846

      goto end;

×

849

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

×

850

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

×

853

  int32_t keyLen = 0;

×

855

  FOREACH(node, group) {

×

856

    SExprNode* pExpr = (SExprNode*)node;

×

857

    keyLen += pExpr->resType.bytes;

×

860

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

×

861

  keyLen += nullFlagSize;

×

863

  keyBuf = taosMemoryCalloc(1, keyLen);

×

864

  if (keyBuf == NULL) {

×

865

    code = terrno;

×

866

    goto end;

×

869

  if (initRemainGroups) {

×

870

    pTableListInfo->remainGroups =

×

871

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

×

872

    if (pTableListInfo->remainGroups == NULL) {

×

873

      code = terrno;

×

874

      goto end;

×

878

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

×

879

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

×

880

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

×

882

    char* isNull = (char*)keyBuf;

×

883

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

×

884

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

×

885

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

×

887

      if (colDataIsNull_s(pValue, i)) {

×

888

        isNull[j] = 1;

×

890

        isNull[j] = 0;

×

891

        char* data = colDataGetData(pValue, i);

×

892

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

×

893

          if (tTagIsJson(data)) {

×

894

            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;

×

895

            goto end;

×

897

          if (tTagIsJsonNull(data)) {

×

898

            isNull[j] = 1;

×

899

            continue;

×

901

          int32_t len = getJsonValueLen(data);

×

902

          memcpy(pStart, data, len);

×

903

          pStart += len;

×

904

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

×

905

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

×

906

            code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;

×

907

            goto end;

×

909

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

×

910

          pStart += varDataTLen(data);

×

912

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

×

913

          pStart += pValue->info.bytes;

×

918

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

×

919

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

×

920

    if (initRemainGroups) {

×

922

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

×

924

      if (code == TSDB_CODE_DUP_KEY) {

×

925

        code = TSDB_CODE_SUCCESS;

×

927

      QUERY_CHECK_CODE(code, lino, end);

×

931

  if (tsTagFilterCache) {

×

932

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

×

933

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

×

935

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

×

937

                                              taosArrayGetSize(tableList) * sizeof(STableKeyInfo));

×

938

    QUERY_CHECK_CODE(code, lino, end);

×

944

end:

×

945

  taosMemoryFreeClear(keyBuf);

×

946

  taosHashCleanup(ctx.colHash);

×

947

  taosArrayDestroy(ctx.cInfoList);

×

948

  blockDataDestroy(pResBlock);

×

949

  taosArrayDestroy(pBlockList);

×

950

  taosArrayDestroyEx(pUidTagList, freeItem);

×

951

  taosArrayDestroyP(groupData, releaseColInfoData);

×

952

  if (code != TSDB_CODE_SUCCESS) {

×

953

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

×

955

  return code;

×

958

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

×

959

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

×

960

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

×

962

  int32_t ret = strcmp(pName1, pName2);

×

963

  if (ret == 0) {

×

964

    return 0;

×

966

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

×

970

static SArray* getTableNameList(const SNodeListNode* pList) {

×

971

  int32_t    code = TSDB_CODE_SUCCESS;

×

972

  int32_t    lino = 0;

×

973

  int32_t    len = LIST_LENGTH(pList->pNodeList);

×

974

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

×

976

  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);

×

977

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

×

979

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

×

980

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

×

981

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

×

982

      terrno = TSDB_CODE_INVALID_PARA;

×

983

      taosArrayDestroy(pTbList);

×

984

      return NULL;

×

987

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

×

988

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

×

989

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

×

990

    cell = cell->pNext;

×

993

  size_t numOfTables = taosArrayGetSize(pTbList);

×

996

  taosArraySort(pTbList, nameComparFn);

×

999

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

×

1000

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

×

1001

  void*   tmpTbl = taosArrayGet(pTbList, 0);

×

1002

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

×

1003

  void*   tmp = taosArrayPush(pNewList, tmpTbl);

×

1004

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

×

1006

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

×

1007

    char** name = taosArrayGetLast(pNewList);

×

1008

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

×

1009

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

×

1010

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

×

1011

      continue;

×

1014

    tmp = taosArrayPush(pNewList, nameInOldList);

×

1015

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

×

1018

_end:

×

1019

  taosArrayDestroy(pTbList);

×

1020

  if (code != TSDB_CODE_SUCCESS) {

×

1021

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

×

1022

    return NULL;

×

1024

  return pNewList;

×

1027

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

×

1028

  uint64_t u1 = *(uint64_t*)a;

×

1029

  uint64_t u2 = *(uint64_t*)b;

×

1031

  if (u1 == u2) {

×

1032

    return 0;

×

1035

  return u1 < u2 ? -1 : 1;

×

1038

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

×

1039

  STUidTagInfo* p1 = (STUidTagInfo*)a;

×

1040

  STUidTagInfo* p2 = (STUidTagInfo*)b;

×

1042

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

×

1043

    return 0;

×

1046

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

×

1049

static FilterCondType checkTagCond(SNode* cond) {

×

1050

  if (nodeType(cond) == QUERY_NODE_OPERATOR) {

×

1051

    return FILTER_NO_LOGIC;

×

1053

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

×

1054

    return FILTER_AND;

×

1056

  return FILTER_OTHER;

×

1059

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

×

1060

  int32_t ret = -1;

×

1061

  int32_t ntype = nodeType(cond);

×

1063

  if (ntype == QUERY_NODE_OPERATOR) {

×

1064

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

×

1067

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

×

1068

    return ret;

×

1071

  bool                 hasTbnameCond = false;

×

1072

  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;

×

1073

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

×

1075

  int32_t len = LIST_LENGTH(pList);

×

1076

  if (len <= 0) {

×

1077

    return ret;

×

1080

  SListCell* cell = pList->pHead;

×

1081

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

×

1082

    if (cell == NULL) break;

×

1083

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

×

1084

      hasTbnameCond = true;

×

1085

      break;

×

1087

    cell = cell->pNext;

×

1090

  taosArraySort(list, filterTableInfoCompare);

×

1091

  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);

×

1093

  if (hasTbnameCond) {

×

1094

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

×

1097

  return ret;

×

1101

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

×

1103

  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {

×

1104

    return -1;

×

1107

  SOperatorNode* pNode = (SOperatorNode*)pTagCond;

×

1108

  if (pNode->opType != OP_TYPE_IN) {

×

1109

    return -1;

×

1112

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

×

1113

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

×

1114

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

×

1115

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

×

1117

    int32_t len = LIST_LENGTH(pList->pNodeList);

×

1118

    if (len <= 0) {

×

1119

      return -1;

×

1122

    SArray*   pTbList = getTableNameList(pList);

×

1123

    int32_t   numOfTables = taosArrayGetSize(pTbList);

×

1124

    SHashObj* uHash = NULL;

×

1126

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

×

1127

    if (numOfExisted > 0) {

×

1128

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

×

1129

      if (!uHash) {

×

1130

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

×

1131

        return terrno;

×

1134

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

×

1135

        STUidTagInfo* pTInfo = taosArrayGet(pExistedUidList, i);

×

1136

        if (!pTInfo) {

×

1137

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

×

1138

          return terrno;

×

1140

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

×

1141

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

×

1142

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

×

1143

          return tempRes;

×

1148

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

×

1149

      char* name = taosArrayGetP(pTbList, i);

×

1151

      uint64_t uid = 0, csuid = 0;

×

1152

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

×

1153

        ETableType tbType = TSDB_TABLE_MAX;

×

1154

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

×

1155

          if (suid != csuid) {

×

1156

            continue;

×

1158

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

×

1159

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

×

1160

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

×

1161

            if (!tmp) {

×

1162

              return terrno;

×

1166

          taosArrayDestroy(pTbList);

×

1167

          taosHashCleanup(uHash);

×

1168

          return -1;

×

1172

        terrno = 0;

×

1176

    taosHashCleanup(uHash);

×

1177

    taosArrayDestroy(pTbList);

×

1178

    return 0;

×

1181

  return -1;

×

1184

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

×

1186

  int32_t      code = TSDB_CODE_SUCCESS;

×

1187

  int32_t      lino = 0;

×

1188

  SSDataBlock* pResBlock = NULL;

×

1189

  code = createDataBlock(&pResBlock);

×

1190

  QUERY_CHECK_CODE(code, lino, _end);

×

1192

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

×

1193

    SColumnInfoData colInfo = {0};

×

1194

    void* tmp = taosArrayGet(pColList, i);

×

1195

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

×

1196

    colInfo.info = *(SColumnInfo*)tmp;

×

1197

    code = blockDataAppendColInfo(pResBlock, &colInfo);

×

1198

    QUERY_CHECK_CODE(code, lino, _end);

×

1201

  code = blockDataEnsureCapacity(pResBlock, numOfTables);

×

1202

  if (code != TSDB_CODE_SUCCESS) {

×

1203

    terrno = code;

×

1204

    blockDataDestroy(pResBlock);

×

1205

    return NULL;

×

1208

  pResBlock->info.rows = numOfTables;

×

1210

  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);

×

1212

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

×

1213

    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);

×

1214

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

×

1216

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

×

1217

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

×

1218

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

×

1220

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

×

1221

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

×

1222

        if (p1->name != NULL) {

×

1223

          STR_TO_VARSTR(str, p1->name);

×

1225

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

×

1226

          QUERY_CHECK_CODE(code, lino, _end);

×

1229

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

×

1230

        QUERY_CHECK_CODE(code, lino, _end);

×

1235

        STagVal tagVal = {0};

×

1236

        tagVal.cid = pColInfo->info.colId;

×

1237

        if (p1->pTagVal == NULL) {

×

1238

          colDataSetNULL(pColInfo, i);

×

1240

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

×

1242

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

×

1243

            colDataSetNULL(pColInfo, i);

×

1244

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

×

1245

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

×

1246

            QUERY_CHECK_CODE(code, lino, _end);

×

1247

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

×

1248

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

×

1249

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

×

1250

            varDataSetLen(tmp, tagVal.nData);

×

1251

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

×

1252

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

×

1256

            taosMemoryFree(tmp);

×

1257

            QUERY_CHECK_CODE(code, lino, _end);

×

1259

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

×

1260

            QUERY_CHECK_CODE(code, lino, _end);

×

1274

_end:

×

1275

  if (code != TSDB_CODE_SUCCESS) {

×

1276

    blockDataDestroy(pResBlock);

×

1277

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

×

1278

    terrno = code;

×

1279

    return NULL;

×

1281

  return pResBlock;

×

1284

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

×

1286

  taosArrayClear(pUidList);

×

1288

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

×

1289

  int32_t       numOfTables = taosArrayGetSize(pUidTagList);

×

1290

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

×

1291

    if (pResultList[i]) {

×

1292

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

×

1293

      if (!tmpTag) {

×

1294

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

×

1295

        return terrno;

×

1297

      uint64_t uid = tmpTag->uid;

×

1298

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

×

1300

      info.uid = uid;

×

1301

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

×

1302

      if (p == NULL) {

×

1303

        return terrno;

×

1306

      if (addUid) {

×

1307

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

×

1308

        if (tmp == NULL) {

×

1309

          return terrno;

×

1315

  return TSDB_CODE_SUCCESS;

×

1318

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

×

1319

  int32_t code = TSDB_CODE_SUCCESS;

×

1320

  int32_t numOfExisted = taosArrayGetSize(pUidList);

×

1321

  if (numOfExisted == 0) {

×

1322

    return code;

×

1325

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

×

1326

    uint64_t*    uid = taosArrayGet(pUidList, i);

×

1327

    if (!uid) {

×

1328

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

×

1329

      return terrno;

×

1331

    STUidTagInfo info = {.uid = *uid};

×

1332

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

×

1333

    if (!tmp) {

×

1334

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

×

1335

      return code;

×

1338

  return code;

×

1341

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

×

1343

  *listAdded = false;

×

1344

  if (pTagCond == NULL) {

×

1345

    return TSDB_CODE_SUCCESS;

×

1348

  terrno = TSDB_CODE_SUCCESS;

×

1350

  int32_t      lino = 0;

×

1351

  int32_t      code = TSDB_CODE_SUCCESS;

×

1352

  SArray*      pBlockList = NULL;

×

1353

  SSDataBlock* pResBlock = NULL;

×

1354

  SScalarParam output = {0};

×

1355

  SArray*      pUidTagList = NULL;

×

1357

  tagFilterAssist ctx = {0};

×

1358

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

×

1359

  if (ctx.colHash == NULL) {

×

1360

    code = terrno;

×

1361

    QUERY_CHECK_CODE(code, lino, end);

×

1364

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

×

1365

  if (ctx.cInfoList == NULL) {

×

1366

    code = terrno;

×

1367

    QUERY_CHECK_CODE(code, lino, end);

×

1370

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

×

1371

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

1372

    terrno = code = ctx.code;

×

1373

    goto end;

×

1376

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

×

1379

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

×

1380

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

×

1382

  code = copyExistedUids(pUidTagList, pUidList);

×

1383

  QUERY_CHECK_CODE(code, lino, end);

×

1385

  FilterCondType condType = checkTagCond(pTagCond);

×

1387

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

×

1388

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

×

1389

    taosArrayClear(pUidList);

×

1391

    int32_t numOfRows = taosArrayGetSize(pUidTagList);

×

1392

    code = taosArrayEnsureCap(pUidList, numOfRows);

×

1393

    QUERY_CHECK_CODE(code, lino, end);

×

1395

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

×

1396

      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);

×

1397

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

×

1398

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

×

1399

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

×

1401

    terrno = 0;

×

1403

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

×

1404

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

×

1406

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

×

1408

    if (code != TSDB_CODE_SUCCESS) {

×

1409

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

×

1410

      terrno = code;

×

1411

      QUERY_CHECK_CODE(code, lino, end);

×

1415

  int32_t numOfTables = taosArrayGetSize(pUidTagList);

×

1416

  if (numOfTables == 0) {

×

1417

    goto end;

×

1420

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

×

1421

  if (pResBlock == NULL) {

×

1422

    code = terrno;

×

1423

    QUERY_CHECK_CODE(code, lino, end);

×

1428

  pBlockList = taosArrayInit(2, POINTER_BYTES);

×

1429

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

×

1431

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

×

1432

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

×

1434

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

×

1435

  if (code != TSDB_CODE_SUCCESS) {

×

1436

    terrno = code;

×

1437

    QUERY_CHECK_CODE(code, lino, end);

×

1440

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

×

1441

  if (code != TSDB_CODE_SUCCESS) {

×

1442

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

×

1443

    terrno = code;

×

1444

    QUERY_CHECK_CODE(code, lino, end);

×

1447

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

×

1448

  if (code != TSDB_CODE_SUCCESS) {

×

1449

    terrno = code;

×

1450

    QUERY_CHECK_CODE(code, lino, end);

×

1452

  *listAdded = true;

×

1454

end:

×

1455

  if (code != TSDB_CODE_SUCCESS) {

×

1456

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

×

1458

  taosHashCleanup(ctx.colHash);

×

1459

  taosArrayDestroy(ctx.cInfoList);

×

1460

  blockDataDestroy(pResBlock);

×

1461

  taosArrayDestroy(pBlockList);

×

1462

  taosArrayDestroyEx(pUidTagList, freeItem);

×

1464

  colDataDestroy(output.columnData);

×

1465

  taosMemoryFreeClear(output.columnData);

×

1466

  return code;

×

1469

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

×

1471

  int32_t code = TSDB_CODE_SUCCESS;

×

1472

  int32_t lino = 0;

×

1473

  size_t  numOfTables = 0;

×

1474

  bool    listAdded = false;

×

1476

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

×

1477

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

×

1479

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

×

1480

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

×

1482

  SIdxFltStatus status = SFLT_NOT_INDEX;

×

1483

  if (pScanNode->tableType != TSDB_SUPER_TABLE) {

×

1484

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

×

1485

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

×

1486

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

×

1487

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

×

1489

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

×

1490

    QUERY_CHECK_CODE(code, lino, _end);

×

1492

    T_MD5_CTX context = {0};

×

1494

    if (tsTagFilterCache) {

×

1496

      code = genTagFilterDigest(pTagCond, &context);

×

1497

      QUERY_CHECK_CODE(code, lino, _error);

×

1499

      bool acquired = false;

×

1500

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

×

1502

      QUERY_CHECK_CODE(code, lino, _error);

×

1504

      if (acquired) {

×

1505

        digest[0] = 1;

×

1506

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

×

1507

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

×

1508

        goto _end;

×

1512

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

×

1513

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

×

1514

      QUERY_CHECK_CODE(code, lino, _error);

×

1517

      if (pTagIndexCond) {

×

1518

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

×

1520

        SIndexMetaArg metaArg = {.metaEx = pVnode,

×

1521

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

×

1523

                                 .suid = pScanNode->uid};

×

1525

        status = SFLT_NOT_INDEX;

×

1526

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

×

1527

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

×

1528

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

×

1530

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

×

1535

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

×

1536

    QUERY_CHECK_CODE(code, lino, _end);

×

1539

    numOfTables = taosArrayGetSize(pUidList);

×

1541

    if (tsTagFilterCache) {

×

1542

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

×

1543

      char*  pPayload = taosMemoryMalloc(size);

×

1544

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

×

1546

      *(int32_t*)pPayload = numOfTables;

×

1547

      if (numOfTables > 0) {

×

1548

        void* tmp = taosArrayGet(pUidList, 0);

×

1549

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

×

1550

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

×

1553

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

×

1555

      QUERY_CHECK_CODE(code, lino, _error);

×

1557

      digest[0] = 1;

×

1558

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

×

1562

_end:

×

1563

  if (!listAdded) {

×

1564

    numOfTables = taosArrayGetSize(pUidList);

×

1565

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

×

1566

      void* tmp = taosArrayGet(pUidList, i);

×

1567

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

×

1568

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

×

1570

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

×

1571

      if (p == NULL) {

×

1572

        taosArrayDestroy(pUidList);

×

1573

        return terrno;

×

1576

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

×

1580

_error:

×

1581

  taosArrayDestroy(pUidList);

×

1582

  if (code != TSDB_CODE_SUCCESS) {

×

1583

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

×

1585

  return code;

×

1588

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

×

1589

  int32_t        code = TSDB_CODE_SUCCESS;

×

1590

  int32_t        lino = 0;

×

1591

  SSubplan*      pSubplan = (SSubplan*)node;

×

1592

  SScanPhysiNode pNode = {0};

×

1593

  pNode.suid = suid;

×

1594

  pNode.uid = suid;

×

1595

  pNode.tableType = TSDB_SUPER_TABLE;

×

1597

  STableListInfo* pTableListInfo = tableListCreate();

×

1598

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

×

1599

  uint8_t         digest[17] = {0};

×

1601

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

×

1603

  QUERY_CHECK_CODE(code, lino, _end);

×

1604

  *tableList = pTableListInfo->pTableList;

×

1605

  pTableListInfo->pTableList = NULL;

×

1606

  tableListDestroy(pTableListInfo);

×

1608

_end:

×

1609

  if (code != TSDB_CODE_SUCCESS) {

×

1610

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

×

1612

  return code;

×

1615

size_t getTableTagsBufLen(const SNodeList* pGroups) {

×

1616

  size_t keyLen = 0;

×

1619

  FOREACH(node, pGroups) {

×

1620

    SExprNode* pExpr = (SExprNode*)node;

×

1621

    keyLen += pExpr->resType.bytes;

×

1624

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

×

1625

  return keyLen;

×

1628

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

×

1630

  SMetaReader mr = {0};

×

1632

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

×

1633

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

×

1634

    pAPI->metaReaderFn.clearReader(&mr);

×

1635

    return TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

1638

  SNodeList* groupNew = NULL;

×

1639

  int32_t    code = nodesCloneList(pGroupNode, &groupNew);

×

1640

  if (TSDB_CODE_SUCCESS != code) {

×

1641

    pAPI->metaReaderFn.clearReader(&mr);

×

1642

    return code;

×

1645

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

×

1646

  nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);

×

1647

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

1648

    nodesDestroyList(groupNew);

×

1649

    pAPI->metaReaderFn.clearReader(&mr);

×

1650

    return code;

×

1652

  char* isNull = (char*)keyBuf;

×

1653

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

×

1656

  int32_t index = 0;

×

1657

  FOREACH(pNode, groupNew) {

×

1658

    SNode*  pNew = NULL;

×

1659

    int32_t code = scalarCalculateConstants(pNode, &pNew);

×

1660

    if (TSDB_CODE_SUCCESS == code) {

×

1661

      REPLACE_NODE(pNew);

×

1663

      nodesDestroyList(groupNew);

×

1664

      pAPI->metaReaderFn.clearReader(&mr);

×

1665

      return code;

×

1668

    if (nodeType(pNew) != QUERY_NODE_VALUE) {

×

1669

      nodesDestroyList(groupNew);

×

1670

      pAPI->metaReaderFn.clearReader(&mr);

×

1671

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

×

1672

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1674

    SValueNode* pValue = (SValueNode*)pNew;

×

1676

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

×

1677

      isNull[index++] = 1;

×

1678

      continue;

×

1680

      isNull[index++] = 0;

×

1681

      char* data = nodesGetValueFromNode(pValue);

×

1682

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

×

1683

        if (tTagIsJson(data)) {

×

1684

          terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;

×

1685

          nodesDestroyList(groupNew);

×

1686

          pAPI->metaReaderFn.clearReader(&mr);

×

1687

          return terrno;

×

1689

        int32_t len = getJsonValueLen(data);

×

1690

        memcpy(pStart, data, len);

×

1691

        pStart += len;

×

1692

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

×

1693

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

×

1694

        pStart += varDataTLen(data);

×

1696

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

×

1697

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

×

1702

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

×

1703

  *pGroupId = calcGroupId(keyBuf, len);

×

1705

  nodesDestroyList(groupNew);

×

1706

  pAPI->metaReaderFn.clearReader(&mr);

×

1708

  return TSDB_CODE_SUCCESS;

×

1711

SArray* makeColumnArrayFromList(SNodeList* pNodeList) {

×

1712

  if (!pNodeList) {

×

1713

    return NULL;

×

1716

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

1717

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

×

1718

  if (pList == NULL) {

×

1719

    return NULL;

×

1722

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

×

1723

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

×

1724

    if (!pColNode) {

×

1725

      taosArrayDestroy(pList);

×

1726

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

×

1727

      return NULL;

×

1731

    SColumn c = {0};

×

1732

    c.slotId = pColNode->slotId;

×

1733

    c.colId = pColNode->colId;

×

1734

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

×

1735

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

×

1736

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

×

1737

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

×

1739

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

×

1740

    if (!tmp) {

×

1741

      taosArrayDestroy(pList);

×

1742

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

×

1743

      return NULL;

×

1747

  return pList;

×

1750

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

×

1752

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

1753

  int32_t code = TSDB_CODE_SUCCESS;

×

1754

  int32_t lino = 0;

×

1756

  pMatchInfo->matchType = type;

×

1758

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

×

1759

  if (pList == NULL) {

×

1760

    code = terrno;

×

1761

    return code;

×

1764

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

×

1765

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

×

1766

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

×

1767

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

×

1768

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

×

1770

      SColMatchItem c = {.needOutput = true};

×

1771

      c.colId = pColNode->colId;

×

1772

      c.srcSlotId = pColNode->slotId;

×

1773

      c.dstSlotId = pNode->slotId;

×

1774

      c.isPk = pColNode->isPk;

×

1775

      c.dataType = pColNode->node.resType;

×

1776

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

×

1777

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

×

1782

  *numOfOutputCols = 0;

×

1783

  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);

×

1784

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

×

1785

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

×

1786

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

×

1790

    if (pNode->slotId >= numOfCols) {

×

1791

      (*numOfOutputCols) += 1;

×

1792

      continue;

×

1795

    SColMatchItem* info = NULL;

×

1796

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

×

1797

      info = taosArrayGet(pList, j);

×

1798

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

×

1799

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

×

1800

        break;

×

1804

    if (pNode->output) {

×

1805

      (*numOfOutputCols) += 1;

×

1806

    } else if (info != NULL) {

×

1808

      info->needOutput = false;

×

1812

  pMatchInfo->pList = pList;

×

1814

_end:

×

1815

  if (code != TSDB_CODE_SUCCESS) {

×

1816

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

×

1818

  return code;

×

1821

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

×

1823

  SResSchema s = {0};

×

1824

  s.scale = scale;

×

1825

  s.type = type;

×

1826

  s.bytes = bytes;

×

1827

  s.slotId = slotId;

×

1828

  s.precision = precision;

×

1829

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

×

1831

  return s;

×

1834

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

×

1835

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

×

1836

  if (pCol == NULL) {

×

1837

    return NULL;

×

1840

  pCol->slotId = slotId;

×

1841

  pCol->colId = colId;

×

1842

  pCol->bytes = pType->bytes;

×

1843

  pCol->type = pType->type;

×

1844

  pCol->scale = pType->scale;

×

1845

  pCol->precision = pType->precision;

×

1846

  pCol->dataBlockId = blockId;

×

1847

  pCol->colType = colType;

×

1848

  return pCol;

×

1851

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

×

1852

  int32_t code = TSDB_CODE_SUCCESS;

×

1853

  int32_t lino = 0;

×

1854

  pExp->base.numOfParams = 0;

×

1855

  pExp->base.pParam = NULL;

×

1856

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

×

1857

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

×

1859

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

×

1860

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

×

1862

  int32_t type = nodeType(pNode);

×

1864

  if (type == QUERY_NODE_COLUMN) {

×

1865

    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;

×

1866

    SColumnNode* pColNode = (SColumnNode*)pNode;

×

1868

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

×

1869

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

×

1871

    pExp->base.numOfParams = 1;

×

1873

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

×

1875

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

×

1877

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

×

1878

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

×

1879

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

×

1881

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

×

1882

  } else if (type == QUERY_NODE_VALUE) {

×

1883

    pExp->pExpr->nodeType = QUERY_NODE_VALUE;

×

1884

    SValueNode* pValNode = (SValueNode*)pNode;

×

1886

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

×

1887

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

×

1889

    pExp->base.numOfParams = 1;

×

1891

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

×

1893

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

×

1894

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

×

1895

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

×

1896

    QUERY_CHECK_CODE(code, lino, _end);

×

1897

  } else if (type == QUERY_NODE_FUNCTION) {

×

1898

    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;

×

1899

    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;

×

1901

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

×

1903

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

×

1904

    tExprNode* pExprNode = pExp->pExpr;

×

1906

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

×

1907

    pExprNode->_function.pFunctNode = pFuncNode;

×

1908

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

×

1910

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

×

1914

    const char* name = "tbname";

×

1915

    int32_t     len = strlen(name);

×

1917

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

×

1918

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

×

1919

      pFuncNode->pParameterList = NULL;

×

1920

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

×

1921

      SValueNode* res = NULL;

×

1922

      if (TSDB_CODE_SUCCESS == code) {

×

1923

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

×

1925

      QUERY_CHECK_CODE(code, lino, _end);

×

1926

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

×

1927

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

×

1928

      if (code != TSDB_CODE_SUCCESS) {

×

1929

        nodesDestroyNode((SNode*)res);

×

1930

        res = NULL;

×

1932

      QUERY_CHECK_CODE(code, lino, _end);

×

1936

    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);

×

1938

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

×

1939

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

×

1940

    pExp->base.numOfParams = numOfParam;

×

1942

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

×

1943

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

×

1944

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

×

1945

      if (p1->type == QUERY_NODE_COLUMN) {

×

1946

        SColumnNode* pcn = (SColumnNode*)p1;

×

1948

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

×

1949

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

×

1950

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

×

1951

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

×

1952

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

×

1953

        SValueNode* pvn = (SValueNode*)p1;

×

1954

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

×

1955

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

×

1956

        QUERY_CHECK_CODE(code, lino, _end);

×

1959

  } else if (type == QUERY_NODE_OPERATOR) {

×

1960

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

1961

    SOperatorNode* pOpNode = (SOperatorNode*)pNode;

×

1963

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

×

1964

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

×

1965

    pExp->base.numOfParams = 1;

×

1967

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

×

1969

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

×

1970

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

×

1971

  } else if (type == QUERY_NODE_CASE_WHEN) {

×

1972

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

1973

    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;

×

1975

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

×

1976

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

×

1977

    pExp->base.numOfParams = 1;

×

1979

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

×

1981

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

×

1982

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

×

1983

  } else if (type == QUERY_NODE_LOGIC_CONDITION) {

×

1984

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

1985

    SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;

×

1986

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

×

1987

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

×

1988

    pExp->base.numOfParams = 1;

×

1989

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

×

1990

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

×

1991

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

×

1993

    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1994

    QUERY_CHECK_CODE(code, lino, _end);

×

1997

_end:

×

1998

  if (code != TSDB_CODE_SUCCESS) {

×

1999

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

×

2001

  return code;

×

2004

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

×

2005

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

×

2008

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

×

2009

  *numOfExprs = LIST_LENGTH(pNodeList);

×

2010

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

×

2011

  if (!pExprs) {

×

2012

    return NULL;

×

2015

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

×

2016

    SExprInfo* pExp = &pExprs[i];

×

2017

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

×

2018

    if (code != TSDB_CODE_SUCCESS) {

×

2019

      taosMemoryFreeClear(pExprs);

×

2020

      terrno = code;

×

2021

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

×

2022

      return NULL;

×

2026

  return pExprs;

×

2029

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

×

2030

  QRY_PARAM_CHECK(pExprInfo);

×

2032

  int32_t code = 0;

×

2033

  int32_t numOfFuncs = LIST_LENGTH(pNodeList);

×

2034

  int32_t numOfGroupKeys = 0;

×

2035

  if (pGroupKeys != NULL) {

×

2036

    numOfGroupKeys = LIST_LENGTH(pGroupKeys);

×

2039

  *numOfExprs = numOfFuncs + numOfGroupKeys;

×

2040

  if (*numOfExprs == 0) {

×

2041

    return code;

×

2044

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

×

2045

  if (pExprs == NULL) {

×

2046

    return terrno;

×

2049

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

×

2050

    STargetNode* pTargetNode = NULL;

×

2051

    if (i < numOfFuncs) {

×

2052

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

×

2054

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

×

2056

    if (!pTargetNode) {

×

2057

      destroyExprInfo(pExprs, *numOfExprs);

×

2058

      taosMemoryFreeClear(pExprs);

×

2059

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

×

2060

      return terrno;

×

2063

    SExprInfo* pExp = &pExprs[i];

×

2064

    code = createExprFromTargetNode(pExp, pTargetNode);

×

2065

    if (code != TSDB_CODE_SUCCESS) {

×

2066

      destroyExprInfo(pExprs, *numOfExprs);

×

2067

      taosMemoryFreeClear(pExprs);

×

2068

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

×

2069

      return code;

×

2073

  *pExprInfo = pExprs;

×

2074

  return code;

×

2078

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

×

2079

  int32_t num = 0;

×

2080

  int32_t code = TSDB_CODE_SUCCESS;

×

2081

  int32_t lino = 0;

×

2083

  SqlFunctionCtx*  p = NULL;

×

2084

  SqlFunctionCtx** pValCtx = taosMemoryCalloc(numOfOutput, POINTER_BYTES);

×

2085

  if (pValCtx == NULL) {

×

2086

    return terrno;

×

2089

  SHashObj* pSelectFuncs = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

2090

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

×

2092

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

×

2093

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

×

2094

    if ((strcmp(pName, "_select_value") == 0) || (strcmp(pName, "_group_key") == 0) ||

×

2095

        (strcmp(pName, "_group_const_value") == 0)) {

×

2096

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

×

2097

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

×

2098

      void* data = taosHashGet(pSelectFuncs, pName, strlen(pName));

×

2099

      if (taosHashGetSize(pSelectFuncs) != 0 && data == NULL) {

×

2100

        p = NULL;

×

2101

        break;

×

2103

        int32_t tempRes = taosHashPut(pSelectFuncs, pName, strlen(pName), &num, sizeof(num));

×

2104

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

×

2105

          code = tempRes;

×

2106

          QUERY_CHECK_CODE(code, lino, _end);

×

2108

        p = &pCtx[i];

×

2112

  taosHashCleanup(pSelectFuncs);

×

2114

  if (p != NULL) {

×

2115

    p->subsidiaries.pCtx = pValCtx;

×

2116

    p->subsidiaries.num = num;

×

2118

    taosMemoryFreeClear(pValCtx);

×

2121

_end:

×

2122

  if (code != TSDB_CODE_SUCCESS) {

×

2123

    taosMemoryFreeClear(pValCtx);

×

2124

    taosHashCleanup(pSelectFuncs);

×

2125

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

×

2127

  return code;

×

2130

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

×

2132

  int32_t         code = TSDB_CODE_SUCCESS;

×

2133

  int32_t         lino = 0;

×

2134

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

×

2135

  if (pFuncCtx == NULL) {

×

2136

    return NULL;

×

2139

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

×

2140

  if (*rowEntryInfoOffset == 0) {

×

2141

    taosMemoryFreeClear(pFuncCtx);

×

2142

    return NULL;

×

2145

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

×

2146

    SExprInfo* pExpr = &pExprInfo[i];

×

2148

    SExprBasicInfo* pFunct = &pExpr->base;

×

2149

    SqlFunctionCtx* pCtx = &pFuncCtx[i];

×

2151

    pCtx->functionId = -1;

×

2152

    pCtx->pExpr = pExpr;

×

2154

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

×

2155

      SFuncExecEnv env = {0};

×

2156

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

×

2157

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

×

2158

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

×

2160

      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);

×

2161

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

×

2162

        if (!isUdaf) {

×

2163

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

×

2164

          QUERY_CHECK_CODE(code, lino, _end);

×

2166

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

×

2167

          pCtx->udfName = taosStrdup(udfName);

×

2168

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

×

2170

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

×

2171

          QUERY_CHECK_CODE(code, lino, _end);

×

2173

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

×

2174

        if (!tmp) {

×

2175

          code = terrno;

×

2176

          QUERY_CHECK_CODE(code, lino, _end);

×

2179

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

×

2180

        if (code != TSDB_CODE_SUCCESS && isUdaf) {

×

2181

          code = TSDB_CODE_SUCCESS;

×

2183

        QUERY_CHECK_CODE(code, lino, _end);

×

2185

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

×

2186

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

×

2187

          if (!tmp) {

×

2188

            code = terrno;

×

2189

            QUERY_CHECK_CODE(code, lino, _end);

×

2193

      pCtx->resDataInfo.interBufSize = env.calcMemSize;

×

2194

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

×

2195

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

×

2197

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

×

2200

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

×

2201

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

×

2202

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

×

2203

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

×

2204

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

×

2206

    pCtx->pTsOutput = NULL;

×

2207

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

×

2208

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

×

2209

    pCtx->order = TSDB_ORDER_ASC;

×

2210

    pCtx->start.key = INT64_MIN;

×

2211

    pCtx->end.key = INT64_MIN;

×

2212

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

×

2213

    pCtx->param = pFunct->pParam;

×

2214

    pCtx->saveHandle.currentPage = -1;

×

2215

    pCtx->pStore = pStore;

×

2216

    pCtx->hasWindowOrGroup = false;

×

2217

    pCtx->needCleanup = false;

×

2220

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

×

2221

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

×

2222

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

×

2225

  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);

×

2226

  QUERY_CHECK_CODE(code, lino, _end);

×

2228

_end:

×

2229

  if (code != TSDB_CODE_SUCCESS) {

×

2230

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

×

2231

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

×

2232

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

×

2233

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

×

2235

    taosMemoryFreeClear(*rowEntryInfoOffset);

×

2236

    taosMemoryFreeClear(pFuncCtx);

×

2238

    terrno = code;

×

2239

    return NULL;

×

2241

  return pFuncCtx;

×

2246

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

×

2247

  int32_t code = TSDB_CODE_SUCCESS;

×

2248

  size_t  numOfSrcCols = taosArrayGetSize(pCols);

×

2250

  int32_t i = 0, j = 0;

×

2251

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

×

2252

    SColumnInfoData* p = taosArrayGet(pCols, i);

×

2253

    if (!p) {

×

2254

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

×

2255

      return terrno;

×

2257

    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);

×

2258

    if (!pmInfo) {

×

2259

      return terrno;

×

2262

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

×

2263

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

×

2264

      if (!pDst) {

×

2265

        return terrno;

×

2267

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

×

2268

      if (code != TSDB_CODE_SUCCESS) {

×

2269

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

×

2270

        return code;

×

2272

      i++;

×

2273

      j++;

×

2274

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

×

2275

      i++;

×

2277

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

×

2278

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

2281

  return code;

×

2284

SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {

×

2285

  SInterval interval = {

×

2286

      .interval = pTableScanNode->interval,

×

2287

      .sliding = pTableScanNode->sliding,

×

2288

      .intervalUnit = pTableScanNode->intervalUnit,

×

2289

      .slidingUnit = pTableScanNode->slidingUnit,

×

2290

      .offset = pTableScanNode->offset,

×

2291

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

×

2294

  calcIntervalAutoOffset(&interval);

×

2296

  return interval;

×

2299

SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {

×

2300

  SColumn c = {0};

×

2302

  c.slotId = pColNode->slotId;

×

2303

  c.colId = pColNode->colId;

×

2304

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

×

2305

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

×

2306

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

×

2307

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

×

2308

  return c;

×

2311

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

×

2313

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

×

2314

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

×

2316

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

×

2317

  if (!pCond->colList) {

×

2318

    return terrno;

×

2320

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

×

2321

  if (pCond->pSlotList == NULL) {

×

2322

    taosMemoryFreeClear(pCond->colList);

×

2323

    return terrno;

×

2327

  pCond->twindows = pTableScanNode->scanRange;

×

2328

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

×

2329

  pCond->type = TIMEWINDOW_RANGE_CONTAINED;

×

2330

  pCond->startVersion = -1;

×

2331

  pCond->endVersion = -1;

×

2332

  pCond->skipRollup = readHandle->skipRollup;

×

2335

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

×

2336

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

×

2338

  int32_t j = 0;

×

2339

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

×

2340

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

×

2341

    if (!pNode) {

×

2342

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

×

2343

      return terrno;

×

2345

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

×

2346

    if (pColNode->colType == COLUMN_TYPE_TAG) {

×

2347

      continue;

×

2350

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

×

2351

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

×

2352

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

×

2353

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

×

2355

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

×

2356

    j += 1;

×

2359

  pCond->numOfCols = j;

×

2360

  return TSDB_CODE_SUCCESS;

×

2363

void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {

×

2364

  taosMemoryFreeClear(pCond->colList);

×

2365

  taosMemoryFreeClear(pCond->pSlotList);

×

2366

×

2368

int32_t convertFillType(int32_t mode) {

×

2369

  int32_t type = TSDB_FILL_NONE;

×

2370

  switch (mode) {

×

2371

    case FILL_MODE_PREV:

×

2372

      type = TSDB_FILL_PREV;

×

2373

      break;

×

2374

    case FILL_MODE_NONE:

×

2375

      type = TSDB_FILL_NONE;

×

2376

      break;

×

2377

    case FILL_MODE_NULL:

×

2378

      type = TSDB_FILL_NULL;

×

2379

      break;

×

2380

    case FILL_MODE_NULL_F:

×

2381

      type = TSDB_FILL_NULL_F;

×

2382

      break;

×

2383

    case FILL_MODE_NEXT:

×

2384

      type = TSDB_FILL_NEXT;

×

2385

      break;

×

2386

    case FILL_MODE_VALUE:

×

2387

      type = TSDB_FILL_SET_VALUE;

×

2388

      break;

×

2389

    case FILL_MODE_VALUE_F:

×

2390

      type = TSDB_FILL_SET_VALUE_F;

×

2391

      break;

×

2392

    case FILL_MODE_LINEAR:

×

2393

      type = TSDB_FILL_LINEAR;

×

2394

      break;

×

2395

    case FILL_MODE_NEAR:

×

2396

      type = TSDB_FILL_NEAR;

×

2397

      break;

×

2398

    default:

×

2399

      type = TSDB_FILL_NONE;

×

2402

  return type;

×

2405

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

×

2406

  if (ascQuery) {

×

2407

    *w = getAlignQueryTimeWindow(pInterval, ts);

×

2410

    *w = getAlignQueryTimeWindow(pInterval, ts);

×

2412

    int64_t key = w->skey;

×

2413

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

×

2414

      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);

×

2415

      if (key > ts) {

×

2416

        break;

×

2419

      w->skey = key;

×

2421

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

×

2423

×

2425

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

×

2426

  STimeWindow w = {0};

×

2428

  w.skey = taosTimeTruncate(ts, pInterval);

×

2429

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

×

2430

  return w;

×

2433

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

×

2434

  STimeWindow win = *pWindow;

×

2435

  STimeWindow save = win;

×

2436

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

×

2437

    save = win;

×

2439

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

×

2442

  return save;

×

2447

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

×

2449

  STimeWindow w = {0};

×

2450

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

×

2451

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

×

2452

    return w;

×

2455

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

×

2456

  if (pRow) {

×

2457

    w = pRow->win;

×

2461

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

×

2462

    w = doCalculateTimeWindow(ts, pInterval);

×

2465

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

×

2468

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

×

2471

  return w;

×

2474

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

×

2475

  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);

×

2476

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

×

2477

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

×

2478

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

×

2479

  return nextStart;

×

2482

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

×

2483

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

×

2484

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

×

2485

×

2487

bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {

×

2488

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

×

2489

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

×

2492

bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {

×

2493

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

×

2496

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

×

2497

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

×

2498

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

×

2500

  pLimitInfo->limit = limit;

×

2501

  pLimitInfo->slimit = slimit;

×

2502

  pLimitInfo->remainOffset = limit.offset;

×

2503

  pLimitInfo->remainGroupOffset = slimit.offset;

×

2504

×

2506

void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {

×

2507

  pLimitInfo->numOfOutputRows = 0;

×

2508

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

×

2509

×

2511

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

×

2512

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

×

2513

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

×

2514

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

2516

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

×

2517

  return TSDB_CODE_SUCCESS;

×

2520

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

×

2522

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

×

2523

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

×

2524

    return NULL;

×

2527

  return taosArrayGet(pTableList->pTableList, index);

×

2530

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

×

2531

  int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);

×

2532

  if (startIndex >= numOfTables) {

×

2533

    return -1;

×

2536

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

×

2537

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

×

2538

    if (!p) {

×

2539

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

×

2540

      return -1;

×

2542

    if (p->uid == uid) {

×

2543

      return i;

×

2546

  return -1;

×

2549

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

×

2550

  *psuid = pTableList->idInfo.suid;

×

2551

  *uid = pTableList->idInfo.uid;

×

2552

  *type = pTableList->idInfo.tableType;

×

2553

×

2555

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

×

2556

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

×

2557

  if (slot == NULL) {

×

2558

    return -1;

×

2561

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

×

2563

  return pKeyInfo->groupId;

×

2567

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

×

2568

  int32_t code = TSDB_CODE_SUCCESS;

×

2569

  int32_t lino = 0;

×

2570

  if (pTableList->map == NULL) {

×

2571

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

×

2572

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

×

2575

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

×

2576

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

×

2577

  if (p != NULL) {

×

2578

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

×

2579

    goto _end;

×

2582

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

×

2583

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

×

2585

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

×

2586

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

×

2587

  if (code != TSDB_CODE_SUCCESS) {

×

2589

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

×

2590

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

×

2593

_end:

×

2594

  if (code != TSDB_CODE_SUCCESS) {

×

2595

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

×

2597

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

×

2600

  return code;

×

2603

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

×

2605

  int32_t totalGroups = tableListGetOutputGroups(pTableList);

×

2606

  int32_t numOfTables = 0;

×

2607

  int32_t code = tableListGetSize(pTableList, &numOfTables);

×

2608

  if (code != TSDB_CODE_SUCCESS) {

×

2609

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

×

2610

    return code;

×

2613

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

×

2614

    return TSDB_CODE_INVALID_PARA;

×

2619

  if (totalGroups == 1) {

×

2620

    *size = numOfTables;

×

2621

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

×

2622

    return TSDB_CODE_SUCCESS;

×

2623

  } else if (totalGroups == numOfTables) {

×

2624

    *size = 1;

×

2625

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

×

2626

    return TSDB_CODE_SUCCESS;

×

2629

  int32_t offset = pTableList->groupOffset[ordinalGroupIndex];

×

2630

  if (ordinalGroupIndex < totalGroups - 1) {

×

2631

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

×

2633

    *size = numOfTables - offset;

×

2636

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

×

2637

  return TSDB_CODE_SUCCESS;

×

2640

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

×

2642

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

×

2644

STableListInfo* tableListCreate() {

×

2645

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

×

2646

  if (pListInfo == NULL) {

×

2647

    return NULL;

×

2650

  pListInfo->remainGroups = NULL;

×

2651

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

×

2652

  if (pListInfo->pTableList == NULL) {

×

2653

    goto _error;

×

2656

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

×

2657

  if (pListInfo->map == NULL) {

×

2658

    goto _error;

×

2661

  pListInfo->numOfOuputGroups = 1;

×

2662

  return pListInfo;

×

2664

_error:

×

2665

  tableListDestroy(pListInfo);

×

2666

  return NULL;

×

2669

void tableListDestroy(STableListInfo* pTableListInfo) {

×

2670

  if (pTableListInfo == NULL) {

×

2671

    return;

×

2674

  taosArrayDestroy(pTableListInfo->pTableList);

×

2675

  taosMemoryFreeClear(pTableListInfo->groupOffset);

×

2677

  taosHashCleanup(pTableListInfo->map);

×

2678

  taosHashCleanup(pTableListInfo->remainGroups);

×

2679

  pTableListInfo->pTableList = NULL;

×

2680

  pTableListInfo->map = NULL;

×

2681

  taosMemoryFree(pTableListInfo);

×

2684

void tableListClear(STableListInfo* pTableListInfo) {

×

2685

  if (pTableListInfo == NULL) {

×

2686

    return;

×

2689

  taosArrayClear(pTableListInfo->pTableList);

×

2690

  taosHashClear(pTableListInfo->map);

×

2691

  taosHashClear(pTableListInfo->remainGroups);

×

2692

  taosMemoryFree(pTableListInfo->groupOffset);

×

2693

  pTableListInfo->numOfOuputGroups = 1;

×

2694

  pTableListInfo->oneTableForEachGroup = false;

×

2697

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

×

2698

  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;

×

2699

  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;

×

2701

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

×

2702

    return 0;

×

2704

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

×

2708

static int32_t sortTableGroup(STableListInfo* pTableListInfo) {

×

2709

  taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);

×

2710

  int32_t size = taosArrayGetSize(pTableListInfo->pTableList);

×

2712

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

×

2713

  if (!pList) {

×

2714

    return terrno;

×

2717

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

×

2718

  if (!pInfo) {

×

2719

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

×

2720

    return terrno;

×

2722

  uint64_t       gid = pInfo->groupId;

×

2724

  int32_t start = 0;

×

2725

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

×

2726

  if (!tmp) {

×

2727

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

×

2728

    return terrno;

×

2731

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

×

2732

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

×

2733

    if (!pInfo) {

×

2734

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

×

2735

      return terrno;

×

2737

    if (pInfo->groupId != gid) {

×

2738

      tmp = taosArrayPush(pList, &i);

×

2739

      if (!tmp) {

×

2740

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

×

2741

        return terrno;

×

2743

      gid = pInfo->groupId;

×

2747

  pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);

×

2748

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

×

2749

  if (pTableListInfo->groupOffset == NULL) {

×

2750

    taosArrayDestroy(pList);

×

2751

    return terrno;

×

2754

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

×

2755

  taosArrayDestroy(pList);

×

2756

  return TSDB_CODE_SUCCESS;

×

2759

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

×

2761

  int32_t code = TSDB_CODE_SUCCESS;

×

2763

  bool   groupByTbname = groupbyTbname(group);

×

2764

  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);

×

2765

  if (!numOfTables) {

×

2766

    return code;

×

2768

  if (group == NULL || groupByTbname) {

×

2769

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

×

2770

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

×

2771

      pTableListInfo->remainGroups =

×

2772

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

×

2773

      if (pTableListInfo->remainGroups == NULL) {

×

2774

        return terrno;

×

2777

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

×

2778

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

×

2779

        if (!info) {

×

2780

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

×

2781

          return terrno;

×

2783

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

×

2785

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

×

2786

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

×

2787

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

×

2788

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

×

2789

          return tempRes;

×

2793

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

×

2794

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

×

2795

        if (!info) {

×

2796

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

×

2797

          return terrno;

×

2799

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

×

2803

    pTableListInfo->oneTableForEachGroup = groupByTbname;

×

2804

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

×

2805

      pTableListInfo->oneTableForEachGroup = true;

×

2808

    if (groupSort && groupByTbname) {

×

2809

      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);

×

2810

      pTableListInfo->numOfOuputGroups = numOfTables;

×

2811

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

×

2812

      pTableListInfo->numOfOuputGroups = numOfTables;

×

2814

      pTableListInfo->numOfOuputGroups = 1;

×

2817

    bool initRemainGroups = false;

×

2818

    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {

×

2819

      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;

×

2820

      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&

×

2821

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

×

2822

        initRemainGroups = true;

×

2826

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

×

2827

    if (code != TSDB_CODE_SUCCESS) {

×

2828

      return code;

×

2831

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

×

2833

    if (groupSort || pScanNode->groupOrderScan) {

×

2834

      code = sortTableGroup(pTableListInfo);

×

2839

  size_t size = taosArrayGetSize(pTableListInfo->pTableList);

×

2840

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

×

2841

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

×

2842

    if (!p) {

×

2843

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

×

2844

      return terrno;

×

2846

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

×

2847

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

×

2848

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

×

2849

      return tempRes;

×

2853

  return code;

×

2856

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

×

2859

  int64_t     st = taosGetTimestampUs();

×

2860

  const char* idStr = GET_TASKID(pTaskInfo);

×

2862

  if (pHandle == NULL) {

×

2863

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

×

2864

    return TSDB_CODE_INVALID_PARA;

×

2867

  uint8_t digest[17] = {0};

×

2868

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

×

2870

  if (code != TSDB_CODE_SUCCESS) {

×

2871

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

×

2872

    return code;

×

2875

  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);

×

2877

  int64_t st1 = taosGetTimestampUs();

×

2878

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

×

2879

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

×

2882

  if (numOfTables == 0) {

×

2883

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

×

2884

    return TSDB_CODE_SUCCESS;

×

2887

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

×

2889

  if (code != TSDB_CODE_SUCCESS) {

×

2890

    return code;

×

2893

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

×

2894

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

×

2896

  return TSDB_CODE_SUCCESS;

×

2899

char* getStreamOpName(uint16_t opType) {

×

2900

  switch (opType) {

×

2901

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:

×

2902

      return "stream scan";

×

2903

    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:

×

2904

      return "project";

×

2905

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:

×

2906

      return "interval single";

×

2907

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:

×

2908

      return "interval final";

×

2909

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:

×

2910

      return "interval semi";

×

2911

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL:

×

2912

      return "interval mid";

×

2913

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:

×

2914

      return "stream fill";

×

2915

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:

×

2916

      return "session single";

×

2917

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION:

×

2918

      return "session semi";

×

2919

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION:

×

2920

      return "session final";

×

2921

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:

×

2922

      return "state single";

×

2923

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:

×

2924

      return "stream partitionby";

×

2925

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT:

×

2926

      return "stream event";

×

2927

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT:

×

2928

      return "stream count";

×

2929

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC:

×

2930

      return "stream interp";

×

2931

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL:

×

2932

      return "interval continue";

×

2934

  return "";

×

2937

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

×

2938

  if (!pBlock) {

×

2939

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

×

2940

    return;

×

2941

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

×

2942

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

×

2943

    return;

×

2945

  if (qDebugFlag & DEBUG_DEBUG) {

×

2946

    char*   pBuf = NULL;

×

2947

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

×

2948

    if (code == 0) {

×

2949

      qDebug("%s", pBuf);

×

2950

      taosMemoryFree(pBuf);

×

2955

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

×

2956

  if (!pBlock) {

×

2957

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

×

2958

    return;

×

2959

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

×

2960

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

×

2963

    return;

×

2965

  if (qDebugFlag & DEBUG_DEBUG) {

×

2966

    char* pBuf = NULL;

×

2968

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

×

2969

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

×

2970

    if (code == 0) {

×

2971

      qDebug("%s", pBuf);

×

2972

      taosMemoryFree(pBuf);

×

2977

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

×

2979

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

×

2980

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

×

2982

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

×

2983

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

×

2984

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

×

2985

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

×

2986

×

2988

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

×

2990

  SColumnDataAgg* pColAgg = NULL;

×

2991

  const char*     isNull = oldkeyBuf;

×

2992

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

×

2994

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

×

2995

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

×

2996

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

×

2997

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

×

2999

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

×

3000

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

×

3002

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

×

3003

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

×

3004

      if (pCol->type == TSDB_DATA_TYPE_JSON) {

×

3005

        int32_t len = getJsonValueLen(val);

×

3006

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

×

3007

        p += len;

×

3008

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

×

3009

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

×

3010

        p += varDataTLen(val);

×

3012

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

×

3013

        p += pCol->bytes;

×

3017

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

×

3018

  return 0;

×

3021

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

×

3022

  uint32_t        colNum = pSortGroupCols->size;

×

3023

  SColumnDataAgg* pColAgg = NULL;

×

3024

  char*           isNull = keyBuf;

×

3025

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

×

3027

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

×

3028

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

×

3029

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

×

3030

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

×

3032

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

×

3034

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

×

3035

      isNull[i] = 1;

×

3037

      isNull[i] = 0;

×

3038

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

×

3039

      if (pCol->type == TSDB_DATA_TYPE_JSON) {

×

3040

        int32_t len = getJsonValueLen(val);

×

3041

        memcpy(p, val, len);

×

3042

        p += len;

×

3043

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

×

3044

        varDataCopy(p, val);

×

3045

        p += varDataTLen(val);

×

3047

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

×

3048

        p += pCol->bytes;

×

3052

  return (int32_t)(p - keyBuf);

×

3055

uint64_t calcGroupId(char* pData, int32_t len) {

×

3057

  tMD5Init(&context);

×

3058

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

×

3059

  tMD5Final(&context);

×

3062

  uint64_t id = 0;

×

3063

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

×

3064

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

×

3065

  return id;

×

3068

SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {

×

3070

  SNodeList* ret = NULL;

×

3071

  FOREACH(node, pSortKeys) {

×

3072

    SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;

×

3073

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

×

3074

    if (code != TSDB_CODE_SUCCESS) {

×

3075

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

×

3076

      terrno = code;

×

3077

      return NULL;

×

3080

  return ret;

×

3083

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

×

3084

  int32_t code = TSDB_CODE_SUCCESS;

×

3085

  int32_t lino = 0;

×

3086

  int32_t len = 0;

×

3087

  int32_t keyNum = taosArrayGetSize(keys);

×

3088

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

×

3089

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

×

3090

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

×

3091

    len += pCol->bytes;

×

3093

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

×

3094

  *pLen = len;

×

3096

_end:

×

3097

  if (code != TSDB_CODE_SUCCESS) {

×

3098

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

×

3100

  return code;

×

taosdata / TDengine / #3599

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

taosdata / TDengine / #3599

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