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

UNCOV

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

×

UNCOV

58

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

×

UNCOV

61

void initResultRowInfo(SResultRowInfo* pResultRowInfo) {

×

UNCOV

62

  pResultRowInfo->size = 0;

×

UNCOV

63

  pResultRowInfo->cur.pageId = -1;

×

UNCOV

64

×

UNCOV

66

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

×

UNCOV

68

void resetResultRow(SResultRow* pResultRow, size_t entrySize) {

×

UNCOV

69

  pResultRow->numOfRows = 0;

×

UNCOV

70

  pResultRow->closed = false;

×

UNCOV

71

  pResultRow->endInterp = false;

×

UNCOV

72

  pResultRow->startInterp = false;

×

UNCOV

74

  if (entrySize > 0) {

×

UNCOV

75

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

×

UNCOV

77

×

UNCOV

80

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

×

UNCOV

81

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

×

UNCOV

84

size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {

×

UNCOV

85

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

×

UNCOV

87

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

×

UNCOV

88

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

×

UNCOV

91

  return rowSize;

×

UNCOV

95

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

×

UNCOV

96

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

×

UNCOV

97

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

×

UNCOV

98

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

100

  SqlFunctionCtx *pCtx = pSup->pCtx;

×

UNCOV

101

  int32_t        *offset = pSup->rowEntryInfoOffset;

×

UNCOV

102

  SResultRow     *pResultRow  = NULL;

×

UNCOV

103

  size_t          processedSize = 0;

×

UNCOV

104

  int32_t         code = TSDB_CODE_SUCCESS;

×

UNCOV

107

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

×

UNCOV

108

  *outBuf = taosMemoryMalloc(*outBufSize);

×

UNCOV

109

  if (*outBuf == NULL) {

×

110

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

×

111

    return terrno;

×

UNCOV

113

  pResultRow = (SResultRow*)*outBuf;

×

UNCOV

114

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

×

UNCOV

115

  inBuf += sizeof(SResultRow);

×

UNCOV

116

  processedSize += sizeof(SResultRow);

×

UNCOV

118

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

×

UNCOV

119

    int32_t len = *(int32_t*)inBuf;

×

UNCOV

120

    inBuf += sizeof(int32_t);

×

UNCOV

121

    processedSize += sizeof(int32_t);

×

UNCOV

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;

×

UNCOV

129

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

×

UNCOV

131

    inBuf += len;

×

UNCOV

132

    processedSize += len;

×

UNCOV

135

  if (processedSize < inBufSize) {

×

UNCOV

137

    size_t leftLen = inBufSize - processedSize;

×

UNCOV

138

    TAOS_MEMORY_REALLOC(*outBuf, *outBufSize + leftLen);

×

UNCOV

139

    if (*outBuf == NULL) {

×

140

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

×

141

      return terrno;

×

UNCOV

143

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

×

UNCOV

144

    inBuf += leftLen;

×

UNCOV

145

    processedSize += leftLen;

×

UNCOV

146

    *outBufSize += leftLen;

×

UNCOV

148

  return TSDB_CODE_SUCCESS;

×

UNCOV

152

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

×

UNCOV

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;

×

UNCOV

158

  SqlFunctionCtx *pCtx = pSup->pCtx;

×

UNCOV

159

  int32_t        *offset = pSup->rowEntryInfoOffset;

×

UNCOV

160

  SResultRow     *pResultRow = (SResultRow*)inBuf;

×

UNCOV

161

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

×

UNCOV

163

  if (rowSize > inBufSize) {

×

164

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

×

165

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

169

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

×

UNCOV

170

  if (rowSize < inBufSize) {

×

UNCOV

171

    *outBufSize += inBufSize - rowSize;

×

UNCOV

174

  *outBuf = taosMemoryMalloc(*outBufSize);

×

UNCOV

175

  if (*outBuf == NULL) {

×

176

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

×

177

    return terrno;

×

UNCOV

180

  char *pBuf = *outBuf;

×

UNCOV

181

  pResultRow->version = FUNCTION_RESULT_INFO_VERSION;

×

UNCOV

182

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

×

UNCOV

183

  pBuf += sizeof(SResultRow);

×

UNCOV

184

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

×

UNCOV

185

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

×

UNCOV

186

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

×

UNCOV

187

    pBuf += sizeof(int32_t);

×

UNCOV

188

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

×

UNCOV

189

    pBuf += len;

×

UNCOV

192

  if (rowSize < inBufSize) {

×

UNCOV

194

    size_t leftLen = inBufSize - rowSize;

×

UNCOV

195

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

×

UNCOV

196

    pBuf += leftLen;

×

UNCOV

198

  return TSDB_CODE_SUCCESS;

×

201

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

×

UNCOV

203

void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {

×

UNCOV

204

  taosMemoryFreeClear(pGroupResInfo->pBuf);

×

UNCOV

205

  if (pGroupResInfo->freeItem) {

×

207

    taosArrayDestroyEx(pGroupResInfo->pRows, freeEx);

×

208

    pGroupResInfo->freeItem = false;

×

209

    pGroupResInfo->pRows = NULL;

×

UNCOV

211

    taosArrayDestroy(pGroupResInfo->pRows);

×

UNCOV

212

    pGroupResInfo->pRows = NULL;

×

UNCOV

214

  pGroupResInfo->index = 0;

×

UNCOV

215

×

UNCOV

217

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

×

UNCOV

218

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

×

UNCOV

219

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

×

UNCOV

221

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

×

UNCOV

222

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

×

UNCOV

223

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

×

UNCOV

225

    if (pts1 == pts2) {

×

226

      return 0;

×

UNCOV

228

      return pts1 < pts2 ? -1 : 1;

×

UNCOV

231

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

×

UNCOV

235

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

×

UNCOV

237

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

×

UNCOV

238

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

239

  int32_t lino = 0;

×

UNCOV

240

  if (pGroupResInfo->pRows != NULL) {

×

UNCOV

241

    taosArrayDestroy(pGroupResInfo->pRows);

×

UNCOV

243

  if (pGroupResInfo->pBuf) {

×

UNCOV

244

    taosMemoryFree(pGroupResInfo->pBuf);

×

UNCOV

245

    pGroupResInfo->pBuf = NULL;

×

UNCOV

249

  int32_t size = tSimpleHashGetSize(pHashmap);

×

UNCOV

251

  void* pData = NULL;

×

UNCOV

252

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

×

UNCOV

253

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

×

UNCOV

255

  size_t  keyLen = 0;

×

UNCOV

256

  int32_t iter = 0;

×

UNCOV

257

  int64_t bufLen = 0, offset = 0;

×

UNCOV

260

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

×

UNCOV

262

    bufLen += keyLen + sizeof(SResultRowPosition);

×

UNCOV

265

  pGroupResInfo->pBuf = taosMemoryMalloc(bufLen);

×

UNCOV

266

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

×

UNCOV

268

  iter = 0;

×

UNCOV

269

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

×

UNCOV

270

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

×

UNCOV

272

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

×

UNCOV

274

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

×

UNCOV

275

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

×

UNCOV

276

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

×

UNCOV

277

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

×

UNCOV

278

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

×

UNCOV

280

    offset += keyLen + sizeof(struct SResultRowPosition);

×

UNCOV

283

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

×

UNCOV

284

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

×

UNCOV

285

    size = POINTER_BYTES;

×

UNCOV

286

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

×

UNCOV

289

  pGroupResInfo->index = 0;

×

UNCOV

291

_end:

×

UNCOV

292

  if (code != TSDB_CODE_SUCCESS) {

×

293

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

×

UNCOV

295

  return code;

×

UNCOV

298

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

×

UNCOV

299

  if (pGroupResInfo->pRows != NULL) {

×

UNCOV

300

    taosArrayDestroy(pGroupResInfo->pRows);

×

UNCOV

303

  pGroupResInfo->freeItem = true;

×

UNCOV

304

  pGroupResInfo->pRows = pArrayList;

×

UNCOV

305

  pGroupResInfo->index = 0;

×

UNCOV

306

×

UNCOV

308

bool hasRemainResults(SGroupResInfo* pGroupResInfo) {

×

UNCOV

309

  if (pGroupResInfo->pRows == NULL) {

×

UNCOV

310

    return false;

×

UNCOV

313

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

×

UNCOV

316

int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) {

×

UNCOV

317

  if (pGroupResInfo->pRows == 0) {

×

UNCOV

318

    return 0;

×

UNCOV

321

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

×

UNCOV

324

SArray* createSortInfo(SNodeList* pNodeList) {

×

UNCOV

325

  size_t numOfCols = 0;

×

UNCOV

327

  if (pNodeList != NULL) {

×

UNCOV

328

    numOfCols = LIST_LENGTH(pNodeList);

×

UNCOV

330

    numOfCols = 0;

×

UNCOV

333

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

×

UNCOV

334

  if (pList == NULL) {

×

335

    return pList;

×

UNCOV

338

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

×

UNCOV

339

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

×

UNCOV

340

    if (!pSortKey) {

×

UNCOV

341

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

×

UNCOV

342

      taosArrayDestroy(pList);

×

343

      pList = NULL;

×

344

      terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

345

      break;

×

UNCOV

347

    SBlockOrderInfo   bi = {0};

×

UNCOV

348

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

×

UNCOV

349

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

×

UNCOV

351

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

×

UNCOV

352

    bi.slotId = pColNode->slotId;

×

UNCOV

353

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

×

UNCOV

354

    if (!tmp) {

×

355

      taosArrayDestroy(pList);

×

356

      pList = NULL;

×

357

      break;

×

UNCOV

361

  return pList;

×

UNCOV

364

SSDataBlock* createDataBlockFromDescNode(SDataBlockDescNode* pNode) {

×

UNCOV

365

  int32_t      numOfCols = LIST_LENGTH(pNode->pSlots);

×

UNCOV

366

  SSDataBlock* pBlock = NULL;

×

UNCOV

367

  int32_t      code = createDataBlock(&pBlock);

×

UNCOV

368

  if (code) {

×

369

    terrno = code;

×

370

    return NULL;

×

UNCOV

373

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

×

UNCOV

374

  pBlock->info.type = STREAM_INVALID;

×

UNCOV

375

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

×

UNCOV

376

  pBlock->info.watermark = INT64_MIN;

×

UNCOV

378

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

×

UNCOV

379

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

×

UNCOV

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;

×

UNCOV

388

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

×

UNCOV

389

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

×

UNCOV

390

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

×

UNCOV

391

    idata.info.noData = pDescNode->reserve;

×

UNCOV

393

    code = blockDataAppendColInfo(pBlock, &idata);

×

UNCOV

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;

×

UNCOV

403

  return pBlock;

×

UNCOV

406

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

×

UNCOV

407

  SDataBlockInfo* pBlockInfo = &pDataBlock->info;

×

UNCOV

409

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

×

UNCOV

410

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

×

UNCOV

411

    if (!pItem) {

×

UNCOV

412

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

×

UNCOV

413

      return terrno;

×

UNCOV

416

    if (pItem->isPk) {

×

UNCOV

417

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

×

UNCOV

418

      if (!pInfoData) {

×

419

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

×

420

        return terrno;

×

UNCOV

422

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

×

UNCOV

423

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

×

UNCOV

426

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

×

UNCOV

427

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

×

UNCOV

428

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

×

429

          return terrno;

×

UNCOV

432

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

×

UNCOV

433

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

×

434

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

×

435

          return terrno;

×

UNCOV

438

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

×

UNCOV

439

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

×

UNCOV

442

      break;

×

UNCOV

446

  return TSDB_CODE_SUCCESS;

×

UNCOV

449

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

×

UNCOV

450

  STransTagExprCtx* pCtx = pContext;

×

UNCOV

451

  SMetaReader*      mr = pCtx->pReader;

×

UNCOV

452

  bool              isTagCol = false, isTbname = false;

×

UNCOV

453

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

×

UNCOV

454

    SColumnNode* pCol = (SColumnNode*)*pNode;

×

UNCOV

455

    if (pCol->colType == COLUMN_TYPE_TBNAME)

×

456

      isTbname = true;

×

UNCOV

458

      isTagCol = true;

×

UNCOV

459

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

×

460

    SFunctionNode* pFunc = (SFunctionNode*)*pNode;

×

461

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

×

UNCOV

463

  if (isTagCol) {

×

UNCOV

464

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

×

UNCOV

466

    SValueNode* res = NULL;

×

UNCOV

467

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

×

UNCOV

468

    if (NULL == res) {

×

469

      return DEAL_RES_ERROR;

×

UNCOV

472

    res->translate = true;

×

UNCOV

473

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

×

UNCOV

475

    STagVal tagVal = {0};

×

UNCOV

476

    tagVal.cid = pSColumnNode->colId;

×

UNCOV

477

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

×

UNCOV

478

    if (p == NULL) {

×

479

      res->node.resType.type = TSDB_DATA_TYPE_NULL;

×

UNCOV

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

×

UNCOV

487

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

×

UNCOV

488

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

×

UNCOV

489

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

×

490

        return DEAL_RES_ERROR;

×

UNCOV

492

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

×

UNCOV

493

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

×

UNCOV

495

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

×

UNCOV

496

      if (code != TSDB_CODE_SUCCESS) {

×

497

        return DEAL_RES_ERROR;

×

UNCOV

500

    nodesDestroyNode(*pNode);

×

UNCOV

501

    *pNode = (SNode*)res;

×

UNCOV

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;

×

UNCOV

523

  return DEAL_RES_CONTINUE;

×

UNCOV

526

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

×

UNCOV

527

  int32_t     code = TSDB_CODE_SUCCESS;

×

UNCOV

528

  SMetaReader mr = {0};

×

UNCOV

530

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

×

UNCOV

531

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

×

UNCOV

532

  if (TSDB_CODE_SUCCESS != code) {

×

533

    pAPI->metaReaderFn.clearReader(&mr);

×

534

    *pQualified = false;

×

536

    return TSDB_CODE_SUCCESS;

×

UNCOV

539

  SNode* pTagCondTmp = NULL;

×

UNCOV

540

  code = nodesCloneNode(pTagCond, &pTagCondTmp);

×

UNCOV

541

  if (TSDB_CODE_SUCCESS != code) {

×

542

    *pQualified = false;

×

543

    pAPI->metaReaderFn.clearReader(&mr);

×

544

    return code;

×

UNCOV

546

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

×

UNCOV

547

  nodesRewriteExprPostOrder(&pTagCondTmp, doTranslateTagExpr, &ctx);

×

UNCOV

548

  pAPI->metaReaderFn.clearReader(&mr);

×

UNCOV

549

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

550

    *pQualified = false;

×

551

    terrno = code;

×

552

    return code;

×

UNCOV

555

  SNode* pNew = NULL;

×

UNCOV

556

  code = scalarCalculateConstants(pTagCondTmp, &pNew);

×

UNCOV

557

  if (TSDB_CODE_SUCCESS != code) {

×

558

    terrno = code;

×

559

    nodesDestroyNode(pTagCondTmp);

×

560

    *pQualified = false;

×

562

    return code;

×

UNCOV

565

  SValueNode* pValue = (SValueNode*)pNew;

×

UNCOV

566

  *pQualified = pValue->datum.b;

×

UNCOV

568

  nodesDestroyNode(pNew);

×

UNCOV

569

  return TSDB_CODE_SUCCESS;

×

UNCOV

572

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

×

UNCOV

573

  tagFilterAssist* pData = (tagFilterAssist*)pContext;

×

UNCOV

574

  SColumnNode*     pSColumnNode = NULL;

×

UNCOV

575

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

×

UNCOV

576

    pSColumnNode = *(SColumnNode**)pNode;

×

UNCOV

577

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

×

UNCOV

578

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

×

UNCOV

579

    if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) {

×

UNCOV

580

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

×

UNCOV

581

      if (NULL == pSColumnNode) {

×

582

        return DEAL_RES_ERROR;

×

UNCOV

584

      pSColumnNode->colId = -1;

×

UNCOV

585

      pSColumnNode->colType = COLUMN_TYPE_TBNAME;

×

UNCOV

586

      pSColumnNode->node.resType.type = TSDB_DATA_TYPE_VARCHAR;

×

UNCOV

587

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

×

UNCOV

588

      nodesDestroyNode(*pNode);

×

UNCOV

589

      *pNode = (SNode*)pSColumnNode;

×

UNCOV

591

      return DEAL_RES_CONTINUE;

×

UNCOV

594

    return DEAL_RES_CONTINUE;

×

UNCOV

597

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

×

UNCOV

598

  if (!data) {

×

UNCOV

600

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

×

UNCOV

601

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

×

602

      return DEAL_RES_ERROR;

×

UNCOV

604

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

×

UNCOV

605

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

×

UNCOV

606

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

×

UNCOV

607

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

×

UNCOV

608

                         .pk = pSColumnNode->isPk};

×

UNCOV

612

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

×

UNCOV

613

    if (!tmp) {

×

614

      return DEAL_RES_ERROR;

×

UNCOV

617

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

×

UNCOV

618

    pSColumnNode->slotId = col->slotId;

×

UNCOV

621

  return DEAL_RES_CONTINUE;

×

UNCOV

624

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

×

UNCOV

625

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

×

UNCOV

626

  if (pColumnData == NULL) {

×

627

    return terrno;

×

UNCOV

630

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

×

UNCOV

631

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

×

UNCOV

632

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

×

UNCOV

633

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

×

UNCOV

635

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

×

UNCOV

636

  if (code != TSDB_CODE_SUCCESS) {

×

637

    terrno = code;

×

638

    releaseColInfoData(pColumnData);

×

639

    return terrno;

×

UNCOV

642

  pParam->columnData = pColumnData;

×

UNCOV

643

  pParam->colAlloced = true;

×

UNCOV

644

  return TSDB_CODE_SUCCESS;

×

UNCOV

647

static void releaseColInfoData(void* pCol) {

×

UNCOV

648

  if (pCol) {

×

UNCOV

649

    SColumnInfoData* col = (SColumnInfoData*)pCol;

×

UNCOV

650

    colDataDestroy(col);

×

UNCOV

651

    taosMemoryFree(col);

×

UNCOV

653

×

UNCOV

655

void freeItem(void* p) {

×

UNCOV

656

  STUidTagInfo* pInfo = p;

×

UNCOV

657

  if (pInfo->pTagVal != NULL) {

×

UNCOV

658

    taosMemoryFree(pInfo->pTagVal);

×

UNCOV

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;

×

UNCOV

710

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

×

UNCOV

712

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

713

  int32_t      lino = 0;

×

UNCOV

714

  SArray*      pBlockList = NULL;

×

UNCOV

715

  SSDataBlock* pResBlock = NULL;

×

UNCOV

716

  void*        keyBuf = NULL;

×

UNCOV

717

  SArray*      groupData = NULL;

×

UNCOV

718

  SArray*      pUidTagList = NULL;

×

UNCOV

719

  SArray*      tableList = NULL;

×

UNCOV

721

  int32_t rows = taosArrayGetSize(pTableListInfo->pTableList);

×

UNCOV

722

  if (rows == 0) {

×

723

    return TSDB_CODE_SUCCESS;

×

UNCOV

726

  tagFilterAssist ctx = {0};

×

UNCOV

727

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

×

UNCOV

728

  if (ctx.colHash == NULL) {

×

729

    code = terrno;

×

730

    goto end;

×

UNCOV

733

  ctx.index = 0;

×

UNCOV

734

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

×

UNCOV

735

  if (ctx.cInfoList == NULL) {

×

736

    code = terrno;

×

737

    goto end;

×

UNCOV

740

  SNode* pNode = NULL;

×

UNCOV

741

  FOREACH(pNode, group) {

×

UNCOV

742

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

×

UNCOV

743

    if (TSDB_CODE_SUCCESS != ctx.code) {

×

744

      code = ctx.code;

×

745

      goto end;

×

UNCOV

747

    REPLACE_NODE(pNode);

×

UNCOV

750

  T_MD5_CTX context = {0};

×

UNCOV

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;

×

UNCOV

776

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

×

UNCOV

777

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

×

UNCOV

779

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

×

UNCOV

780

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

×

UNCOV

781

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

×

UNCOV

782

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

×

UNCOV

783

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

×

UNCOV

784

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

×

UNCOV

787

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

×

UNCOV

788

  if (code != TSDB_CODE_SUCCESS) {

×

789

    goto end;

×

UNCOV

792

  int32_t numOfTables = taosArrayGetSize(pUidTagList);

×

UNCOV

793

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

×

UNCOV

794

  if (pResBlock == NULL) {

×

795

    code = terrno;

×

796

    goto end;

×

UNCOV

802

  pBlockList = taosArrayInit(2, POINTER_BYTES);

×

UNCOV

803

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

×

UNCOV

805

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

×

UNCOV

806

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

×

UNCOV

808

  groupData = taosArrayInit(2, POINTER_BYTES);

×

UNCOV

809

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

×

UNCOV

811

  FOREACH(pNode, group) {

×

UNCOV

812

    SScalarParam output = {0};

×

UNCOV

814

    switch (nodeType(pNode)) {

×

815

      case QUERY_NODE_VALUE:

×

816

        break;

×

UNCOV

817

      case QUERY_NODE_COLUMN:

×

UNCOV

820

        SExprNode* expNode = (SExprNode*)pNode;

×

UNCOV

821

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

×

UNCOV

822

        if (code != TSDB_CODE_SUCCESS) {

×

823

          goto end;

×

UNCOV

825

        break;

×

828

      default:

×

829

        code = TSDB_CODE_OPS_NOT_SUPPORT;

×

830

        goto end;

×

UNCOV

833

    if (nodeType(pNode) == QUERY_NODE_COLUMN) {

×

UNCOV

834

      SColumnNode*     pSColumnNode = (SColumnNode*)pNode;

×

UNCOV

835

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

×

UNCOV

836

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

×

UNCOV

837

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

×

UNCOV

838

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

×

839

      continue;

×

UNCOV

841

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

×

UNCOV

844

    if (code != TSDB_CODE_SUCCESS) {

×

845

      releaseColInfoData(output.columnData);

×

846

      goto end;

×

UNCOV

849

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

×

UNCOV

850

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

×

UNCOV

853

  int32_t keyLen = 0;

×

UNCOV

855

  FOREACH(node, group) {

×

UNCOV

856

    SExprNode* pExpr = (SExprNode*)node;

×

UNCOV

857

    keyLen += pExpr->resType.bytes;

×

UNCOV

860

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

×

UNCOV

861

  keyLen += nullFlagSize;

×

UNCOV

863

  keyBuf = taosMemoryCalloc(1, keyLen);

×

UNCOV

864

  if (keyBuf == NULL) {

×

865

    code = terrno;

×

866

    goto end;

×

UNCOV

869

  if (initRemainGroups) {

×

UNCOV

870

    pTableListInfo->remainGroups =

×

UNCOV

871

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

×

UNCOV

872

    if (pTableListInfo->remainGroups == NULL) {

×

873

      code = terrno;

×

874

      goto end;

×

UNCOV

878

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

×

UNCOV

879

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

×

UNCOV

880

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

×

UNCOV

882

    char* isNull = (char*)keyBuf;

×

UNCOV

883

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

×

UNCOV

884

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

×

UNCOV

885

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

×

UNCOV

887

      if (colDataIsNull_s(pValue, i)) {

×

UNCOV

888

        isNull[j] = 1;

×

UNCOV

890

        isNull[j] = 0;

×

UNCOV

891

        char* data = colDataGetData(pValue, i);

×

UNCOV

892

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

×

UNCOV

893

          if (tTagIsJson(data)) {

×

UNCOV

894

            code = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;

×

UNCOV

895

            goto end;

×

UNCOV

897

          if (tTagIsJsonNull(data)) {

×

898

            isNull[j] = 1;

×

899

            continue;

×

UNCOV

901

          int32_t len = getJsonValueLen(data);

×

UNCOV

902

          memcpy(pStart, data, len);

×

UNCOV

903

          pStart += len;

×

UNCOV

904

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

×

UNCOV

905

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

×

UNCOV

906

            code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;

×

UNCOV

907

            goto end;

×

UNCOV

909

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

×

UNCOV

910

          pStart += varDataTLen(data);

×

UNCOV

912

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

×

UNCOV

913

          pStart += pValue->info.bytes;

×

UNCOV

918

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

×

UNCOV

919

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

×

UNCOV

920

    if (initRemainGroups) {

×

UNCOV

922

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

×

UNCOV

924

      if (code == TSDB_CODE_DUP_KEY) {

×

UNCOV

925

        code = TSDB_CODE_SUCCESS;

×

UNCOV

927

      QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

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

×

UNCOV

944

end:

×

UNCOV

945

  taosMemoryFreeClear(keyBuf);

×

UNCOV

946

  taosHashCleanup(ctx.colHash);

×

UNCOV

947

  taosArrayDestroy(ctx.cInfoList);

×

UNCOV

948

  blockDataDestroy(pResBlock);

×

UNCOV

949

  taosArrayDestroy(pBlockList);

×

UNCOV

950

  taosArrayDestroyEx(pUidTagList, freeItem);

×

UNCOV

951

  taosArrayDestroyP(groupData, releaseColInfoData);

×

UNCOV

952

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

953

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

×

UNCOV

955

  return code;

×

UNCOV

958

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

×

UNCOV

959

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

×

UNCOV

960

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

×

UNCOV

962

  int32_t ret = strcmp(pName1, pName2);

×

UNCOV

963

  if (ret == 0) {

×

UNCOV

964

    return 0;

×

UNCOV

966

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

×

UNCOV

970

static SArray* getTableNameList(const SNodeListNode* pList) {

×

UNCOV

971

  int32_t    code = TSDB_CODE_SUCCESS;

×

UNCOV

972

  int32_t    lino = 0;

×

UNCOV

973

  int32_t    len = LIST_LENGTH(pList->pNodeList);

×

UNCOV

974

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

×

UNCOV

976

  SArray* pTbList = taosArrayInit(len, POINTER_BYTES);

×

UNCOV

977

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

×

UNCOV

979

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

×

UNCOV

980

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

×

UNCOV

981

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

×

982

      terrno = TSDB_CODE_INVALID_PARA;

×

983

      taosArrayDestroy(pTbList);

×

984

      return NULL;

×

UNCOV

987

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

×

UNCOV

988

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

×

UNCOV

989

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

×

UNCOV

990

    cell = cell->pNext;

×

UNCOV

993

  size_t numOfTables = taosArrayGetSize(pTbList);

×

UNCOV

996

  taosArraySort(pTbList, nameComparFn);

×

UNCOV

999

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

×

UNCOV

1000

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

×

UNCOV

1001

  void*   tmpTbl = taosArrayGet(pTbList, 0);

×

UNCOV

1002

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

×

UNCOV

1003

  void*   tmp = taosArrayPush(pNewList, tmpTbl);

×

UNCOV

1004

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

×

UNCOV

1006

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

×

UNCOV

1007

    char** name = taosArrayGetLast(pNewList);

×

UNCOV

1008

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

×

UNCOV

1009

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

×

UNCOV

1010

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

×

UNCOV

1011

      continue;

×

UNCOV

1014

    tmp = taosArrayPush(pNewList, nameInOldList);

×

UNCOV

1015

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

×

UNCOV

1018

_end:

×

UNCOV

1019

  taosArrayDestroy(pTbList);

×

UNCOV

1020

  if (code != TSDB_CODE_SUCCESS) {

×

1021

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

×

1022

    return NULL;

×

UNCOV

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;

×

UNCOV

1038

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

×

UNCOV

1039

  STUidTagInfo* p1 = (STUidTagInfo*)a;

×

UNCOV

1040

  STUidTagInfo* p2 = (STUidTagInfo*)b;

×

UNCOV

1042

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

×

1043

    return 0;

×

UNCOV

1046

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

×

UNCOV

1049

static FilterCondType checkTagCond(SNode* cond) {

×

UNCOV

1050

  if (nodeType(cond) == QUERY_NODE_OPERATOR) {

×

UNCOV

1051

    return FILTER_NO_LOGIC;

×

UNCOV

1053

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

×

UNCOV

1054

    return FILTER_AND;

×

UNCOV

1056

  return FILTER_OTHER;

×

UNCOV

1059

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

×

UNCOV

1060

  int32_t ret = -1;

×

UNCOV

1061

  int32_t ntype = nodeType(cond);

×

UNCOV

1063

  if (ntype == QUERY_NODE_OPERATOR) {

×

UNCOV

1064

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

×

UNCOV

1067

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

×

UNCOV

1068

    return ret;

×

UNCOV

1071

  bool                 hasTbnameCond = false;

×

UNCOV

1072

  SLogicConditionNode* pNode = (SLogicConditionNode*)cond;

×

UNCOV

1073

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

×

UNCOV

1075

  int32_t len = LIST_LENGTH(pList);

×

UNCOV

1076

  if (len <= 0) {

×

1077

    return ret;

×

UNCOV

1080

  SListCell* cell = pList->pHead;

×

UNCOV

1081

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

×

UNCOV

1082

    if (cell == NULL) break;

×

UNCOV

1083

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

×

UNCOV

1084

      hasTbnameCond = true;

×

UNCOV

1085

      break;

×

UNCOV

1087

    cell = cell->pNext;

×

UNCOV

1090

  taosArraySort(list, filterTableInfoCompare);

×

UNCOV

1091

  taosArrayRemoveDuplicate(list, filterTableInfoCompare, NULL);

×

UNCOV

1093

  if (hasTbnameCond) {

×

UNCOV

1094

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

×

UNCOV

1097

  return ret;

×

UNCOV

1101

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

×

UNCOV

1103

  if (nodeType(pTagCond) != QUERY_NODE_OPERATOR) {

×

UNCOV

1104

    return -1;

×

UNCOV

1107

  SOperatorNode* pNode = (SOperatorNode*)pTagCond;

×

UNCOV

1108

  if (pNode->opType != OP_TYPE_IN) {

×

UNCOV

1109

    return -1;

×

UNCOV

1112

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

×

UNCOV

1113

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

×

UNCOV

1114

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

×

UNCOV

1115

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

×

UNCOV

1117

    int32_t len = LIST_LENGTH(pList->pNodeList);

×

UNCOV

1118

    if (len <= 0) {

×

1119

      return -1;

×

UNCOV

1122

    SArray*   pTbList = getTableNameList(pList);

×

UNCOV

1123

    int32_t   numOfTables = taosArrayGetSize(pTbList);

×

UNCOV

1124

    SHashObj* uHash = NULL;

×

UNCOV

1126

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

×

UNCOV

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;

×

UNCOV

1148

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

×

UNCOV

1149

      char* name = taosArrayGetP(pTbList, i);

×

UNCOV

1151

      uint64_t uid = 0, csuid = 0;

×

UNCOV

1152

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

×

UNCOV

1153

        ETableType tbType = TSDB_TABLE_MAX;

×

UNCOV

1154

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

×

UNCOV

1155

          if (suid != csuid) {

×

UNCOV

1156

            continue;

×

UNCOV

1158

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

×

UNCOV

1159

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

×

UNCOV

1160

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

×

UNCOV

1161

            if (!tmp) {

×

1162

              return terrno;

×

UNCOV

1166

          taosArrayDestroy(pTbList);

×

UNCOV

1167

          taosHashCleanup(uHash);

×

UNCOV

1168

          return -1;

×

UNCOV

1172

        terrno = 0;

×

UNCOV

1176

    taosHashCleanup(uHash);

×

UNCOV

1177

    taosArrayDestroy(pTbList);

×

UNCOV

1178

    return 0;

×

UNCOV

1181

  return -1;

×

UNCOV

1184

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

×

UNCOV

1186

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

1187

  int32_t      lino = 0;

×

UNCOV

1188

  SSDataBlock* pResBlock = NULL;

×

UNCOV

1189

  code = createDataBlock(&pResBlock);

×

UNCOV

1190

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1192

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

×

UNCOV

1193

    SColumnInfoData colInfo = {0};

×

UNCOV

1194

    void* tmp = taosArrayGet(pColList, i);

×

UNCOV

1195

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

×

UNCOV

1196

    colInfo.info = *(SColumnInfo*)tmp;

×

UNCOV

1197

    code = blockDataAppendColInfo(pResBlock, &colInfo);

×

UNCOV

1198

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1201

  code = blockDataEnsureCapacity(pResBlock, numOfTables);

×

UNCOV

1202

  if (code != TSDB_CODE_SUCCESS) {

×

1203

    terrno = code;

×

1204

    blockDataDestroy(pResBlock);

×

1205

    return NULL;

×

UNCOV

1208

  pResBlock->info.rows = numOfTables;

×

UNCOV

1210

  int32_t numOfCols = taosArrayGetSize(pResBlock->pDataBlock);

×

UNCOV

1212

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

×

UNCOV

1213

    STUidTagInfo* p1 = taosArrayGet(pUidTagList, i);

×

UNCOV

1214

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

×

UNCOV

1216

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

×

UNCOV

1217

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

×

UNCOV

1218

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

×

UNCOV

1220

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

×

UNCOV

1221

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

×

UNCOV

1222

        if (p1->name != NULL) {

×

UNCOV

1223

          STR_TO_VARSTR(str, p1->name);

×

UNCOV

1225

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

×

UNCOV

1226

          QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1229

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

×

UNCOV

1230

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1235

        STagVal tagVal = {0};

×

UNCOV

1236

        tagVal.cid = pColInfo->info.colId;

×

UNCOV

1237

        if (p1->pTagVal == NULL) {

×

1238

          colDataSetNULL(pColInfo, i);

×

UNCOV

1240

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

×

UNCOV

1242

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

×

UNCOV

1243

            colDataSetNULL(pColInfo, i);

×

UNCOV

1244

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

×

UNCOV

1245

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

×

UNCOV

1246

            QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1247

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

×

UNCOV

1248

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

×

UNCOV

1249

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

×

UNCOV

1250

            varDataSetLen(tmp, tagVal.nData);

×

UNCOV

1251

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

×

UNCOV

1252

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

×

UNCOV

1256

            taosMemoryFree(tmp);

×

UNCOV

1257

            QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1259

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

×

UNCOV

1260

            QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1274

_end:

×

UNCOV

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;

×

UNCOV

1281

  return pResBlock;

×

UNCOV

1284

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

×

UNCOV

1286

  taosArrayClear(pUidList);

×

UNCOV

1288

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

×

UNCOV

1289

  int32_t       numOfTables = taosArrayGetSize(pUidTagList);

×

UNCOV

1290

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

×

UNCOV

1291

    if (pResultList[i]) {

×

UNCOV

1292

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

×

UNCOV

1293

      if (!tmpTag) {

×

1294

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

×

1295

        return terrno;

×

UNCOV

1297

      uint64_t uid = tmpTag->uid;

×

UNCOV

1298

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

×

UNCOV

1300

      info.uid = uid;

×

UNCOV

1301

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

×

UNCOV

1302

      if (p == NULL) {

×

1303

        return terrno;

×

UNCOV

1306

      if (addUid) {

×

1307

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

×

1308

        if (tmp == NULL) {

×

1309

          return terrno;

×

UNCOV

1315

  return TSDB_CODE_SUCCESS;

×

UNCOV

1318

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

×

UNCOV

1319

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1320

  int32_t numOfExisted = taosArrayGetSize(pUidList);

×

UNCOV

1321

  if (numOfExisted == 0) {

×

UNCOV

1322

    return code;

×

UNCOV

1325

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

×

UNCOV

1326

    uint64_t*    uid = taosArrayGet(pUidList, i);

×

UNCOV

1327

    if (!uid) {

×

1328

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

×

1329

      return terrno;

×

UNCOV

1331

    STUidTagInfo info = {.uid = *uid};

×

UNCOV

1332

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

×

UNCOV

1333

    if (!tmp) {

×

1334

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

×

1335

      return code;

×

UNCOV

1338

  return code;

×

UNCOV

1341

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

×

UNCOV

1343

  *listAdded = false;

×

UNCOV

1344

  if (pTagCond == NULL) {

×

UNCOV

1345

    return TSDB_CODE_SUCCESS;

×

UNCOV

1348

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

1350

  int32_t      lino = 0;

×

UNCOV

1351

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

1352

  SArray*      pBlockList = NULL;

×

UNCOV

1353

  SSDataBlock* pResBlock = NULL;

×

UNCOV

1354

  SScalarParam output = {0};

×

UNCOV

1355

  SArray*      pUidTagList = NULL;

×

UNCOV

1357

  tagFilterAssist ctx = {0};

×

UNCOV

1358

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

×

UNCOV

1359

  if (ctx.colHash == NULL) {

×

1360

    code = terrno;

×

1361

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1364

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

×

UNCOV

1365

  if (ctx.cInfoList == NULL) {

×

1366

    code = terrno;

×

1367

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1370

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

×

UNCOV

1371

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

1372

    terrno = code = ctx.code;

×

1373

    goto end;

×

UNCOV

1376

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

×

UNCOV

1379

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

×

UNCOV

1380

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

×

UNCOV

1382

  code = copyExistedUids(pUidTagList, pUidList);

×

UNCOV

1383

  QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1385

  FilterCondType condType = checkTagCond(pTagCond);

×

UNCOV

1387

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

×

UNCOV

1388

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

×

UNCOV

1389

    taosArrayClear(pUidList);

×

UNCOV

1391

    int32_t numOfRows = taosArrayGetSize(pUidTagList);

×

UNCOV

1392

    code = taosArrayEnsureCap(pUidList, numOfRows);

×

UNCOV

1393

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1395

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

×

UNCOV

1396

      STUidTagInfo* pInfo = taosArrayGet(pUidTagList, i);

×

UNCOV

1397

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

×

UNCOV

1398

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

×

UNCOV

1399

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

×

UNCOV

1401

    terrno = 0;

×

UNCOV

1403

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

×

UNCOV

1404

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

×

UNCOV

1406

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

×

UNCOV

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

×

UNCOV

1415

  int32_t numOfTables = taosArrayGetSize(pUidTagList);

×

UNCOV

1416

  if (numOfTables == 0) {

×

UNCOV

1417

    goto end;

×

UNCOV

1420

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

×

UNCOV

1421

  if (pResBlock == NULL) {

×

1422

    code = terrno;

×

1423

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1428

  pBlockList = taosArrayInit(2, POINTER_BYTES);

×

UNCOV

1429

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

×

UNCOV

1431

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

×

UNCOV

1432

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

×

UNCOV

1434

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

×

UNCOV

1435

  if (code != TSDB_CODE_SUCCESS) {

×

1436

    terrno = code;

×

1437

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1440

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

×

UNCOV

1441

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1442

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

×

UNCOV

1443

    terrno = code;

×

UNCOV

1444

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1447

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

×

UNCOV

1448

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1449

    terrno = code;

×

1450

    QUERY_CHECK_CODE(code, lino, end);

×

UNCOV

1452

  *listAdded = true;

×

UNCOV

1454

end:

×

UNCOV

1455

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1456

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

×

UNCOV

1458

  taosHashCleanup(ctx.colHash);

×

UNCOV

1459

  taosArrayDestroy(ctx.cInfoList);

×

UNCOV

1460

  blockDataDestroy(pResBlock);

×

UNCOV

1461

  taosArrayDestroy(pBlockList);

×

UNCOV

1462

  taosArrayDestroyEx(pUidTagList, freeItem);

×

UNCOV

1464

  colDataDestroy(output.columnData);

×

UNCOV

1465

  taosMemoryFreeClear(output.columnData);

×

UNCOV

1466

  return code;

×

UNCOV

1469

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

×

UNCOV

1471

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1472

  int32_t lino = 0;

×

UNCOV

1473

  size_t  numOfTables = 0;

×

UNCOV

1474

  bool    listAdded = false;

×

UNCOV

1476

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

×

UNCOV

1477

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

×

UNCOV

1479

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

×

UNCOV

1480

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

×

UNCOV

1482

  SIdxFltStatus status = SFLT_NOT_INDEX;

×

UNCOV

1483

  if (pScanNode->tableType != TSDB_SUPER_TABLE) {

×

UNCOV

1484

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

×

UNCOV

1485

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

×

UNCOV

1486

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

×

UNCOV

1487

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

×

UNCOV

1489

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

×

UNCOV

1490

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1492

    T_MD5_CTX context = {0};

×

UNCOV

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;

×

UNCOV

1512

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

×

UNCOV

1513

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

×

UNCOV

1514

      QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

1517

      if (pTagIndexCond) {

×

UNCOV

1518

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

×

UNCOV

1520

        SIndexMetaArg metaArg = {.metaEx = pVnode,

×

UNCOV

1521

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

×

UNCOV

1523

                                 .suid = pScanNode->uid};

×

UNCOV

1525

        status = SFLT_NOT_INDEX;

×

UNCOV

1526

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

×

UNCOV

1527

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

×

UNCOV

1528

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

×

UNCOV

1530

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

×

UNCOV

1535

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

×

UNCOV

1536

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1539

    numOfTables = taosArrayGetSize(pUidList);

×

UNCOV

1541

    if (tsTagFilterCache) {

×

UNCOV

1542

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

×

UNCOV

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

×

UNCOV

1562

_end:

×

UNCOV

1563

  if (!listAdded) {

×

UNCOV

1564

    numOfTables = taosArrayGetSize(pUidList);

×

UNCOV

1565

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

×

UNCOV

1566

      void* tmp = taosArrayGet(pUidList, i);

×

UNCOV

1567

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

×

UNCOV

1568

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

×

UNCOV

1570

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

×

UNCOV

1571

      if (p == NULL) {

×

1572

        taosArrayDestroy(pUidList);

×

1573

        return terrno;

×

UNCOV

1576

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

×

UNCOV

1580

_error:

×

UNCOV

1581

  taosArrayDestroy(pUidList);

×

UNCOV

1582

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1583

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

×

UNCOV

1585

  return code;

×

UNCOV

1588

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

×

UNCOV

1589

  int32_t        code = TSDB_CODE_SUCCESS;

×

UNCOV

1590

  int32_t        lino = 0;

×

UNCOV

1591

  SSubplan*      pSubplan = (SSubplan*)node;

×

UNCOV

1592

  SScanPhysiNode pNode = {0};

×

UNCOV

1593

  pNode.suid = suid;

×

UNCOV

1594

  pNode.uid = suid;

×

UNCOV

1595

  pNode.tableType = TSDB_SUPER_TABLE;

×

UNCOV

1597

  STableListInfo* pTableListInfo = tableListCreate();

×

UNCOV

1598

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

×

UNCOV

1599

  uint8_t         digest[17] = {0};

×

UNCOV

1601

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

×

UNCOV

1603

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1604

  *tableList = pTableListInfo->pTableList;

×

UNCOV

1605

  pTableListInfo->pTableList = NULL;

×

UNCOV

1606

  tableListDestroy(pTableListInfo);

×

UNCOV

1608

_end:

×

UNCOV

1609

  if (code != TSDB_CODE_SUCCESS) {

×

1610

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

×

UNCOV

1612

  return code;

×

UNCOV

1615

size_t getTableTagsBufLen(const SNodeList* pGroups) {

×

UNCOV

1616

  size_t keyLen = 0;

×

UNCOV

1619

  FOREACH(node, pGroups) {

×

UNCOV

1620

    SExprNode* pExpr = (SExprNode*)node;

×

UNCOV

1621

    keyLen += pExpr->resType.bytes;

×

UNCOV

1624

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

×

UNCOV

1625

  return keyLen;

×

UNCOV

1628

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

×

UNCOV

1630

  SMetaReader mr = {0};

×

UNCOV

1632

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

×

UNCOV

1633

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

×

1634

    pAPI->metaReaderFn.clearReader(&mr);

×

1635

    return TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

UNCOV

1638

  SNodeList* groupNew = NULL;

×

UNCOV

1639

  int32_t    code = nodesCloneList(pGroupNode, &groupNew);

×

UNCOV

1640

  if (TSDB_CODE_SUCCESS != code) {

×

1641

    pAPI->metaReaderFn.clearReader(&mr);

×

1642

    return code;

×

UNCOV

1645

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

×

UNCOV

1646

  nodesRewriteExprsPostOrder(groupNew, doTranslateTagExpr, &ctx);

×

UNCOV

1647

  if (TSDB_CODE_SUCCESS != ctx.code) {

×

1648

    nodesDestroyList(groupNew);

×

1649

    pAPI->metaReaderFn.clearReader(&mr);

×

1650

    return code;

×

UNCOV

1652

  char* isNull = (char*)keyBuf;

×

UNCOV

1653

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

×

UNCOV

1656

  int32_t index = 0;

×

UNCOV

1657

  FOREACH(pNode, groupNew) {

×

UNCOV

1658

    SNode*  pNew = NULL;

×

UNCOV

1659

    int32_t code = scalarCalculateConstants(pNode, &pNew);

×

UNCOV

1660

    if (TSDB_CODE_SUCCESS == code) {

×

UNCOV

1661

      REPLACE_NODE(pNew);

×

1663

      nodesDestroyList(groupNew);

×

1664

      pAPI->metaReaderFn.clearReader(&mr);

×

1665

      return code;

×

UNCOV

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;

×

UNCOV

1674

    SValueNode* pValue = (SValueNode*)pNew;

×

UNCOV

1676

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

×

1677

      isNull[index++] = 1;

×

1678

      continue;

×

UNCOV

1680

      isNull[index++] = 0;

×

UNCOV

1681

      char* data = nodesGetValueFromNode(pValue);

×

UNCOV

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;

×

UNCOV

1692

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

×

1693

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

×

1694

        pStart += varDataTLen(data);

×

UNCOV

1696

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

×

UNCOV

1697

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

×

UNCOV

1702

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

×

UNCOV

1703

  *pGroupId = calcGroupId(keyBuf, len);

×

UNCOV

1705

  nodesDestroyList(groupNew);

×

UNCOV

1706

  pAPI->metaReaderFn.clearReader(&mr);

×

UNCOV

1708

  return TSDB_CODE_SUCCESS;

×

UNCOV

1711

SArray* makeColumnArrayFromList(SNodeList* pNodeList) {

×

UNCOV

1712

  if (!pNodeList) {

×

1713

    return NULL;

×

UNCOV

1716

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

UNCOV

1717

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

×

UNCOV

1718

  if (pList == NULL) {

×

1719

    return NULL;

×

UNCOV

1722

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

×

UNCOV

1723

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

×

UNCOV

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;

×

UNCOV

1731

    SColumn c = {0};

×

UNCOV

1732

    c.slotId = pColNode->slotId;

×

UNCOV

1733

    c.colId = pColNode->colId;

×

UNCOV

1734

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

×

UNCOV

1735

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

×

UNCOV

1736

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

×

UNCOV

1737

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

×

UNCOV

1739

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

×

UNCOV

1740

    if (!tmp) {

×

1741

      taosArrayDestroy(pList);

×

1742

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

×

1743

      return NULL;

×

UNCOV

1747

  return pList;

×

UNCOV

1750

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

×

UNCOV

1752

  size_t  numOfCols = LIST_LENGTH(pNodeList);

×

UNCOV

1753

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1754

  int32_t lino = 0;

×

UNCOV

1756

  pMatchInfo->matchType = type;

×

UNCOV

1758

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

×

UNCOV

1759

  if (pList == NULL) {

×

1760

    code = terrno;

×

1761

    return code;

×

UNCOV

1764

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

×

UNCOV

1765

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

×

UNCOV

1766

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

×

UNCOV

1767

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

×

UNCOV

1768

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

×

UNCOV

1770

      SColMatchItem c = {.needOutput = true};

×

UNCOV

1771

      c.colId = pColNode->colId;

×

UNCOV

1772

      c.srcSlotId = pColNode->slotId;

×

UNCOV

1773

      c.dstSlotId = pNode->slotId;

×

UNCOV

1774

      c.isPk = pColNode->isPk;

×

UNCOV

1775

      c.dataType = pColNode->node.resType;

×

UNCOV

1776

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

×

UNCOV

1777

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

×

UNCOV

1782

  *numOfOutputCols = 0;

×

UNCOV

1783

  int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);

×

UNCOV

1784

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

×

UNCOV

1785

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

×

UNCOV

1786

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

×

UNCOV

1790

    if (pNode->slotId >= numOfCols) {

×

UNCOV

1791

      (*numOfOutputCols) += 1;

×

UNCOV

1792

      continue;

×

UNCOV

1795

    SColMatchItem* info = NULL;

×

UNCOV

1796

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

×

UNCOV

1797

      info = taosArrayGet(pList, j);

×

UNCOV

1798

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

×

UNCOV

1799

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

×

UNCOV

1800

        break;

×

UNCOV

1804

    if (pNode->output) {

×

UNCOV

1805

      (*numOfOutputCols) += 1;

×

UNCOV

1806

    } else if (info != NULL) {

×

UNCOV

1808

      info->needOutput = false;

×

UNCOV

1812

  pMatchInfo->pList = pList;

×

UNCOV

1814

_end:

×

UNCOV

1815

  if (code != TSDB_CODE_SUCCESS) {

×

1816

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

×

UNCOV

1818

  return code;

×

UNCOV

1821

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

×

UNCOV

1823

  SResSchema s = {0};

×

UNCOV

1824

  s.scale = scale;

×

UNCOV

1825

  s.type = type;

×

UNCOV

1826

  s.bytes = bytes;

×

UNCOV

1827

  s.slotId = slotId;

×

UNCOV

1828

  s.precision = precision;

×

UNCOV

1829

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

×

UNCOV

1831

  return s;

×

UNCOV

1834

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

×

UNCOV

1835

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

×

UNCOV

1836

  if (pCol == NULL) {

×

1837

    return NULL;

×

UNCOV

1840

  pCol->slotId = slotId;

×

UNCOV

1841

  pCol->colId = colId;

×

UNCOV

1842

  pCol->bytes = pType->bytes;

×

UNCOV

1843

  pCol->type = pType->type;

×

UNCOV

1844

  pCol->scale = pType->scale;

×

UNCOV

1845

  pCol->precision = pType->precision;

×

UNCOV

1846

  pCol->dataBlockId = blockId;

×

UNCOV

1847

  pCol->colType = colType;

×

UNCOV

1848

  return pCol;

×

UNCOV

1851

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

×

UNCOV

1852

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1853

  int32_t lino = 0;

×

UNCOV

1854

  pExp->base.numOfParams = 0;

×

UNCOV

1855

  pExp->base.pParam = NULL;

×

UNCOV

1856

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

×

UNCOV

1857

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

×

UNCOV

1859

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

×

UNCOV

1860

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

×

UNCOV

1862

  int32_t type = nodeType(pNode);

×

UNCOV

1864

  if (type == QUERY_NODE_COLUMN) {

×

UNCOV

1865

    pExp->pExpr->nodeType = QUERY_NODE_COLUMN;

×

UNCOV

1866

    SColumnNode* pColNode = (SColumnNode*)pNode;

×

UNCOV

1868

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

×

UNCOV

1869

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

×

UNCOV

1871

    pExp->base.numOfParams = 1;

×

UNCOV

1873

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

×

UNCOV

1875

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

×

UNCOV

1877

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

×

UNCOV

1878

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

×

UNCOV

1879

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

×

UNCOV

1881

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

×

UNCOV

1882

  } else if (type == QUERY_NODE_VALUE) {

×

UNCOV

1883

    pExp->pExpr->nodeType = QUERY_NODE_VALUE;

×

UNCOV

1884

    SValueNode* pValNode = (SValueNode*)pNode;

×

UNCOV

1886

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

×

UNCOV

1887

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

×

UNCOV

1889

    pExp->base.numOfParams = 1;

×

UNCOV

1891

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

×

UNCOV

1893

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

×

UNCOV

1894

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

×

UNCOV

1895

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

×

UNCOV

1896

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1897

  } else if (type == QUERY_NODE_FUNCTION) {

×

UNCOV

1898

    pExp->pExpr->nodeType = QUERY_NODE_FUNCTION;

×

UNCOV

1899

    SFunctionNode* pFuncNode = (SFunctionNode*)pNode;

×

UNCOV

1901

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

×

UNCOV

1903

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

×

UNCOV

1904

    tExprNode* pExprNode = pExp->pExpr;

×

UNCOV

1906

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

×

UNCOV

1907

    pExprNode->_function.pFunctNode = pFuncNode;

×

UNCOV

1908

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

×

UNCOV

1910

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

×

UNCOV

1914

    const char* name = "tbname";

×

UNCOV

1915

    int32_t     len = strlen(name);

×

UNCOV

1917

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

×

UNCOV

1918

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

×

UNCOV

1919

      pFuncNode->pParameterList = NULL;

×

UNCOV

1920

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

×

UNCOV

1921

      SValueNode* res = NULL;

×

UNCOV

1922

      if (TSDB_CODE_SUCCESS == code) {

×

UNCOV

1923

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

×

UNCOV

1925

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1926

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

×

UNCOV

1927

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

×

UNCOV

1928

      if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

1929

        nodesDestroyNode((SNode*)res);

×

1930

        res = NULL;

×

UNCOV

1932

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1936

    int32_t numOfParam = LIST_LENGTH(pFuncNode->pParameterList);

×

UNCOV

1938

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

×

UNCOV

1939

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

×

UNCOV

1940

    pExp->base.numOfParams = numOfParam;

×

UNCOV

1942

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

×

UNCOV

1943

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

×

UNCOV

1944

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

×

UNCOV

1945

      if (p1->type == QUERY_NODE_COLUMN) {

×

UNCOV

1946

        SColumnNode* pcn = (SColumnNode*)p1;

×

UNCOV

1948

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

×

UNCOV

1949

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

×

UNCOV

1950

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

×

UNCOV

1951

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

×

UNCOV

1952

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

×

UNCOV

1953

        SValueNode* pvn = (SValueNode*)p1;

×

UNCOV

1954

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

×

UNCOV

1955

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

×

UNCOV

1956

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1959

  } else if (type == QUERY_NODE_OPERATOR) {

×

UNCOV

1960

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

UNCOV

1961

    SOperatorNode* pOpNode = (SOperatorNode*)pNode;

×

UNCOV

1963

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

×

UNCOV

1964

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

×

UNCOV

1965

    pExp->base.numOfParams = 1;

×

UNCOV

1967

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

×

UNCOV

1969

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

×

UNCOV

1970

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

×

UNCOV

1971

  } else if (type == QUERY_NODE_CASE_WHEN) {

×

UNCOV

1972

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

UNCOV

1973

    SCaseWhenNode* pCaseNode = (SCaseWhenNode*)pNode;

×

UNCOV

1975

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

×

UNCOV

1976

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

×

UNCOV

1977

    pExp->base.numOfParams = 1;

×

UNCOV

1979

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

×

UNCOV

1981

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

×

UNCOV

1982

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

×

UNCOV

1983

  } else if (type == QUERY_NODE_LOGIC_CONDITION) {

×

UNCOV

1984

    pExp->pExpr->nodeType = QUERY_NODE_OPERATOR;

×

UNCOV

1985

    SLogicConditionNode* pCond = (SLogicConditionNode*)pNode;

×

UNCOV

1986

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

×

UNCOV

1987

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

×

UNCOV

1988

    pExp->base.numOfParams = 1;

×

UNCOV

1989

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

×

UNCOV

1990

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

×

UNCOV

1991

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

×

UNCOV

1993

    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

UNCOV

1994

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1997

_end:

×

UNCOV

1998

  if (code != TSDB_CODE_SUCCESS) {

×

1999

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

×

UNCOV

2001

  return code;

×

UNCOV

2004

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

×

UNCOV

2005

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

×

UNCOV

2008

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

×

UNCOV

2009

  *numOfExprs = LIST_LENGTH(pNodeList);

×

UNCOV

2010

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

×

UNCOV

2011

  if (!pExprs) {

×

2012

    return NULL;

×

UNCOV

2015

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

×

UNCOV

2016

    SExprInfo* pExp = &pExprs[i];

×

UNCOV

2017

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

×

UNCOV

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;

×

UNCOV

2026

  return pExprs;

×

UNCOV

2029

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

×

UNCOV

2030

  QRY_PARAM_CHECK(pExprInfo);

×

UNCOV

2032

  int32_t code = 0;

×

UNCOV

2033

  int32_t numOfFuncs = LIST_LENGTH(pNodeList);

×

UNCOV

2034

  int32_t numOfGroupKeys = 0;

×

UNCOV

2035

  if (pGroupKeys != NULL) {

×

UNCOV

2036

    numOfGroupKeys = LIST_LENGTH(pGroupKeys);

×

UNCOV

2039

  *numOfExprs = numOfFuncs + numOfGroupKeys;

×

UNCOV

2040

  if (*numOfExprs == 0) {

×

UNCOV

2041

    return code;

×

UNCOV

2044

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

×

UNCOV

2045

  if (pExprs == NULL) {

×

2046

    return terrno;

×

UNCOV

2049

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

×

UNCOV

2050

    STargetNode* pTargetNode = NULL;

×

UNCOV

2051

    if (i < numOfFuncs) {

×

UNCOV

2052

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

×

UNCOV

2054

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

×

UNCOV

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;

×

UNCOV

2063

    SExprInfo* pExp = &pExprs[i];

×

UNCOV

2064

    code = createExprFromTargetNode(pExp, pTargetNode);

×

UNCOV

2065

    if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

2066

      destroyExprInfo(pExprs, *numOfExprs);

×

2067

      taosMemoryFreeClear(pExprs);

×

2068

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

×

2069

      return code;

×

UNCOV

2073

  *pExprInfo = pExprs;

×

UNCOV

2074

  return code;

×

UNCOV

2078

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

×

UNCOV

2079

  int32_t num = 0;

×

UNCOV

2080

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

2081

  int32_t lino = 0;

×

UNCOV

2083

  SqlFunctionCtx*  p = NULL;

×

UNCOV

2084

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

×

UNCOV

2085

  if (pValCtx == NULL) {

×

2086

    return terrno;

×

UNCOV

2089

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

×

UNCOV

2090

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

×

UNCOV

2092

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

×

UNCOV

2093

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

×

UNCOV

2094

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

×

UNCOV

2095

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

×

UNCOV

2096

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

×

UNCOV

2097

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

×

UNCOV

2098

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

×

UNCOV

2099

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

×

UNCOV

2100

        p = NULL;

×

UNCOV

2101

        break;

×

UNCOV

2103

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

×

UNCOV

2104

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

×

2105

          code = tempRes;

×

2106

          QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2108

        p = &pCtx[i];

×

UNCOV

2112

  taosHashCleanup(pSelectFuncs);

×

UNCOV

2114

  if (p != NULL) {

×

UNCOV

2115

    p->subsidiaries.pCtx = pValCtx;

×

UNCOV

2116

    p->subsidiaries.num = num;

×

UNCOV

2118

    taosMemoryFreeClear(pValCtx);

×

2121

_end:

×

UNCOV

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

×

UNCOV

2127

  return code;

×

UNCOV

2130

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

×

UNCOV

2132

  int32_t         code = TSDB_CODE_SUCCESS;

×

UNCOV

2133

  int32_t         lino = 0;

×

UNCOV

2134

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

×

UNCOV

2135

  if (pFuncCtx == NULL) {

×

2136

    return NULL;

×

UNCOV

2139

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

×

UNCOV

2140

  if (*rowEntryInfoOffset == 0) {

×

2141

    taosMemoryFreeClear(pFuncCtx);

×

2142

    return NULL;

×

UNCOV

2145

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

×

UNCOV

2146

    SExprInfo* pExpr = &pExprInfo[i];

×

UNCOV

2148

    SExprBasicInfo* pFunct = &pExpr->base;

×

UNCOV

2149

    SqlFunctionCtx* pCtx = &pFuncCtx[i];

×

UNCOV

2151

    pCtx->functionId = -1;

×

UNCOV

2152

    pCtx->pExpr = pExpr;

×

UNCOV

2154

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

×

UNCOV

2155

      SFuncExecEnv env = {0};

×

UNCOV

2156

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

×

UNCOV

2157

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

×

UNCOV

2158

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

×

UNCOV

2160

      bool isUdaf = fmIsUserDefinedFunc(pCtx->functionId);

×

UNCOV

2161

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

×

UNCOV

2162

        if (!isUdaf) {

×

UNCOV

2163

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

×

UNCOV

2164

          QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2166

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

×

UNCOV

2167

          pCtx->udfName = taosStrdup(udfName);

×

UNCOV

2168

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

×

UNCOV

2170

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

×

UNCOV

2171

          QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2173

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

×

UNCOV

2174

        if (!tmp) {

×

2175

          code = terrno;

×

2176

          QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2179

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

×

UNCOV

2180

        if (code != TSDB_CODE_SUCCESS && isUdaf) {

×

UNCOV

2181

          code = TSDB_CODE_SUCCESS;

×

UNCOV

2183

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2185

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

×

UNCOV

2186

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

×

UNCOV

2187

          if (!tmp) {

×

UNCOV

2188

            code = terrno;

×

2189

            QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2193

      pCtx->resDataInfo.interBufSize = env.calcMemSize;

×

UNCOV

2194

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

×

UNCOV

2195

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

×

UNCOV

2197

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

×

UNCOV

2200

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

×

UNCOV

2201

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

×

UNCOV

2202

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

×

UNCOV

2203

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

×

UNCOV

2204

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

×

UNCOV

2206

    pCtx->pTsOutput = NULL;

×

UNCOV

2207

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

×

UNCOV

2208

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

×

UNCOV

2209

    pCtx->order = TSDB_ORDER_ASC;

×

UNCOV

2210

    pCtx->start.key = INT64_MIN;

×

UNCOV

2211

    pCtx->end.key = INT64_MIN;

×

UNCOV

2212

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

×

UNCOV

2213

    pCtx->param = pFunct->pParam;

×

UNCOV

2214

    pCtx->saveHandle.currentPage = -1;

×

UNCOV

2215

    pCtx->pStore = pStore;

×

UNCOV

2216

    pCtx->hasWindowOrGroup = false;

×

UNCOV

2217

    pCtx->needCleanup = false;

×

UNCOV

2220

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

×

UNCOV

2221

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

×

UNCOV

2222

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

×

UNCOV

2225

  code = setSelectValueColumnInfo(pFuncCtx, numOfOutput);

×

UNCOV

2226

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

2228

_end:

×

UNCOV

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;

×

UNCOV

2241

  return pFuncCtx;

×

UNCOV

2246

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

×

UNCOV

2247

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

2248

  size_t  numOfSrcCols = taosArrayGetSize(pCols);

×

UNCOV

2250

  int32_t i = 0, j = 0;

×

UNCOV

2251

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

×

UNCOV

2252

    SColumnInfoData* p = taosArrayGet(pCols, i);

×

UNCOV

2253

    if (!p) {

×

2254

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

×

2255

      return terrno;

×

UNCOV

2257

    SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, j);

×

UNCOV

2258

    if (!pmInfo) {

×

2259

      return terrno;

×

UNCOV

2262

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

×

UNCOV

2263

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

×

UNCOV

2264

      if (!pDst) {

×

2265

        return terrno;

×

UNCOV

2267

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

×

UNCOV

2268

      if (code != TSDB_CODE_SUCCESS) {

×

2269

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

×

2270

        return code;

×

UNCOV

2272

      i++;

×

UNCOV

2273

      j++;

×

UNCOV

2274

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

×

UNCOV

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;

×

UNCOV

2281

  return code;

×

UNCOV

2284

SInterval extractIntervalInfo(const STableScanPhysiNode* pTableScanNode) {

×

UNCOV

2285

  SInterval interval = {

×

UNCOV

2286

      .interval = pTableScanNode->interval,

×

UNCOV

2287

      .sliding = pTableScanNode->sliding,

×

UNCOV

2288

      .intervalUnit = pTableScanNode->intervalUnit,

×

UNCOV

2289

      .slidingUnit = pTableScanNode->slidingUnit,

×

UNCOV

2290

      .offset = pTableScanNode->offset,

×

UNCOV

2291

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

×

UNCOV

2294

  calcIntervalAutoOffset(&interval);

×

UNCOV

2296

  return interval;

×

UNCOV

2299

SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {

×

UNCOV

2300

  SColumn c = {0};

×

UNCOV

2302

  c.slotId = pColNode->slotId;

×

UNCOV

2303

  c.colId = pColNode->colId;

×

UNCOV

2304

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

×

UNCOV

2305

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

×

UNCOV

2306

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

×

UNCOV

2307

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

×

UNCOV

2308

  return c;

×

UNCOV

2311

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

×

UNCOV

2313

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

×

UNCOV

2314

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

×

UNCOV

2316

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

×

UNCOV

2317

  if (!pCond->colList) {

×

2318

    return terrno;

×

UNCOV

2320

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

×

UNCOV

2321

  if (pCond->pSlotList == NULL) {

×

2322

    taosMemoryFreeClear(pCond->colList);

×

2323

    return terrno;

×

UNCOV

2327

  pCond->twindows = pTableScanNode->scanRange;

×

UNCOV

2328

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

×

UNCOV

2329

  pCond->type = TIMEWINDOW_RANGE_CONTAINED;

×

UNCOV

2330

  pCond->startVersion = -1;

×

UNCOV

2331

  pCond->endVersion = -1;

×

UNCOV

2332

  pCond->skipRollup = readHandle->skipRollup;

×

UNCOV

2335

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

×

UNCOV

2336

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

×

UNCOV

2338

  int32_t j = 0;

×

UNCOV

2339

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

×

UNCOV

2340

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

×

UNCOV

2341

    if (!pNode) {

×

UNCOV

2342

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

×

UNCOV

2343

      return terrno;

×

UNCOV

2345

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

×

UNCOV

2346

    if (pColNode->colType == COLUMN_TYPE_TAG) {

×

2347

      continue;

×

UNCOV

2350

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

×

UNCOV

2351

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

×

UNCOV

2352

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

×

UNCOV

2353

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

×

UNCOV

2355

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

×

UNCOV

2356

    j += 1;

×

UNCOV

2359

  pCond->numOfCols = j;

×

UNCOV

2360

  return TSDB_CODE_SUCCESS;

×

UNCOV

2363

void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {

×

UNCOV

2364

  taosMemoryFreeClear(pCond->colList);

×

UNCOV

2365

  taosMemoryFreeClear(pCond->pSlotList);

×

UNCOV

2366

×

UNCOV

2368

int32_t convertFillType(int32_t mode) {

×

UNCOV

2369

  int32_t type = TSDB_FILL_NONE;

×

UNCOV

2370

  switch (mode) {

×

UNCOV

2371

    case FILL_MODE_PREV:

×

UNCOV

2372

      type = TSDB_FILL_PREV;

×

UNCOV

2373

      break;

×

2374

    case FILL_MODE_NONE:

×

2375

      type = TSDB_FILL_NONE;

×

2376

      break;

×

UNCOV

2377

    case FILL_MODE_NULL:

×

UNCOV

2378

      type = TSDB_FILL_NULL;

×

UNCOV

2379

      break;

×

UNCOV

2380

    case FILL_MODE_NULL_F:

×

UNCOV

2381

      type = TSDB_FILL_NULL_F;

×

UNCOV

2382

      break;

×

UNCOV

2383

    case FILL_MODE_NEXT:

×

UNCOV

2384

      type = TSDB_FILL_NEXT;

×

UNCOV

2385

      break;

×

UNCOV

2386

    case FILL_MODE_VALUE:

×

UNCOV

2387

      type = TSDB_FILL_SET_VALUE;

×

UNCOV

2388

      break;

×

UNCOV

2389

    case FILL_MODE_VALUE_F:

×

UNCOV

2390

      type = TSDB_FILL_SET_VALUE_F;

×

UNCOV

2391

      break;

×

UNCOV

2392

    case FILL_MODE_LINEAR:

×

UNCOV

2393

      type = TSDB_FILL_LINEAR;

×

UNCOV

2394

      break;

×

UNCOV

2395

    case FILL_MODE_NEAR:

×

UNCOV

2396

      type = TSDB_FILL_NEAR;

×

UNCOV

2397

      break;

×

2398

    default:

×

2399

      type = TSDB_FILL_NONE;

×

UNCOV

2402

  return type;

×

UNCOV

2405

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

×

UNCOV

2406

  if (ascQuery) {

×

UNCOV

2407

    *w = getAlignQueryTimeWindow(pInterval, ts);

×

UNCOV

2410

    *w = getAlignQueryTimeWindow(pInterval, ts);

×

UNCOV

2412

    int64_t key = w->skey;

×

UNCOV

2413

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

×

UNCOV

2414

      key = getNextTimeWindowStart(pInterval, key, TSDB_ORDER_ASC);

×

UNCOV

2415

      if (key > ts) {

×

UNCOV

2416

        break;

×

UNCOV

2419

      w->skey = key;

×

UNCOV

2421

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

×

UNCOV

2423

×

UNCOV

2425

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

×

UNCOV

2426

  STimeWindow w = {0};

×

UNCOV

2428

  w.skey = taosTimeTruncate(ts, pInterval);

×

UNCOV

2429

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

×

UNCOV

2430

  return w;

×

UNCOV

2433

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

×

UNCOV

2434

  STimeWindow win = *pWindow;

×

UNCOV

2435

  STimeWindow save = win;

×

UNCOV

2436

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

×

UNCOV

2437

    save = win;

×

UNCOV

2439

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

×

UNCOV

2442

  return save;

×

UNCOV

2447

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

×

UNCOV

2449

  STimeWindow w = {0};

×

UNCOV

2450

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

×

UNCOV

2451

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

×

UNCOV

2452

    return w;

×

UNCOV

2455

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

×

UNCOV

2456

  if (pRow) {

×

UNCOV

2457

    w = pRow->win;

×

UNCOV

2461

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

×

UNCOV

2462

    w = doCalculateTimeWindow(ts, pInterval);

×

UNCOV

2465

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

×

UNCOV

2468

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

×

UNCOV

2471

  return w;

×

UNCOV

2474

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

×

UNCOV

2475

  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);

×

UNCOV

2476

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

×

UNCOV

2477

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

×

UNCOV

2478

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

×

UNCOV

2479

  return nextStart;

×

UNCOV

2482

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

×

UNCOV

2483

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

×

UNCOV

2484

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

×

UNCOV

2485

×

UNCOV

2487

bool hasLimitOffsetInfo(SLimitInfo* pLimitInfo) {

×

UNCOV

2488

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

×

UNCOV

2489

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

×

2492

bool hasSlimitOffsetInfo(SLimitInfo* pLimitInfo) {

×

2493

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

×

UNCOV

2496

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

×

UNCOV

2497

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

×

UNCOV

2498

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

×

UNCOV

2500

  pLimitInfo->limit = limit;

×

UNCOV

2501

  pLimitInfo->slimit = slimit;

×

UNCOV

2502

  pLimitInfo->remainOffset = limit.offset;

×

UNCOV

2503

  pLimitInfo->remainGroupOffset = slimit.offset;

×

UNCOV

2504

×

UNCOV

2506

void resetLimitInfoForNextGroup(SLimitInfo* pLimitInfo) {

×

UNCOV

2507

  pLimitInfo->numOfOutputRows = 0;

×

UNCOV

2508

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

×

UNCOV

2509

×

UNCOV

2511

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

×

UNCOV

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;

×

UNCOV

2516

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

×

UNCOV

2517

  return TSDB_CODE_SUCCESS;

×

UNCOV

2520

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

×

UNCOV

2522

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

×

UNCOV

2523

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

×

UNCOV

2524

    return NULL;

×

UNCOV

2527

  return taosArrayGet(pTableList->pTableList, index);

×

UNCOV

2530

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

×

UNCOV

2531

  int32_t numOfTables = taosArrayGetSize(pTableList->pTableList);

×

UNCOV

2532

  if (startIndex >= numOfTables) {

×

2533

    return -1;

×

UNCOV

2536

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

×

UNCOV

2537

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

×

UNCOV

2538

    if (!p) {

×

2539

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

×

2540

      return -1;

×

UNCOV

2542

    if (p->uid == uid) {

×

UNCOV

2543

      return i;

×

UNCOV

2546

  return -1;

×

UNCOV

2549

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

×

UNCOV

2550

  *psuid = pTableList->idInfo.suid;

×

UNCOV

2551

  *uid = pTableList->idInfo.uid;

×

UNCOV

2552

  *type = pTableList->idInfo.tableType;

×

UNCOV

2553

×

UNCOV

2555

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

×

UNCOV

2556

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

×

UNCOV

2557

  if (slot == NULL) {

×

2558

    return -1;

×

UNCOV

2561

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

×

UNCOV

2563

  return pKeyInfo->groupId;

×

UNCOV

2567

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

×

UNCOV

2568

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

2569

  int32_t lino = 0;

×

UNCOV

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

×

UNCOV

2575

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

×

UNCOV

2576

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

×

UNCOV

2577

  if (p != NULL) {

×

UNCOV

2578

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

×

UNCOV

2579

    goto _end;

×

UNCOV

2582

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

×

UNCOV

2583

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

×

UNCOV

2585

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

×

UNCOV

2586

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

×

UNCOV

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

×

UNCOV

2593

_end:

×

UNCOV

2594

  if (code != TSDB_CODE_SUCCESS) {

×

2595

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

×

UNCOV

2597

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

×

UNCOV

2600

  return code;

×

UNCOV

2603

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

×

UNCOV

2605

  int32_t totalGroups = tableListGetOutputGroups(pTableList);

×

UNCOV

2606

  int32_t numOfTables = 0;

×

UNCOV

2607

  int32_t code = tableListGetSize(pTableList, &numOfTables);

×

UNCOV

2608

  if (code != TSDB_CODE_SUCCESS) {

×

2609

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

×

2610

    return code;

×

UNCOV

2613

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

×

2614

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

2619

  if (totalGroups == 1) {

×

UNCOV

2620

    *size = numOfTables;

×

UNCOV

2621

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

×

UNCOV

2622

    return TSDB_CODE_SUCCESS;

×

2623

  } else if (totalGroups == numOfTables) {

×

UNCOV

2624

    *size = 1;

×

UNCOV

2625

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

×

UNCOV

2626

    return TSDB_CODE_SUCCESS;

×

2629

  int32_t offset = pTableList->groupOffset[ordinalGroupIndex];

×

2630

  if (ordinalGroupIndex < totalGroups - 1) {

×

UNCOV

2631

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

×

2633

    *size = numOfTables - offset;

×

2636

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

×

UNCOV

2637

  return TSDB_CODE_SUCCESS;

×

UNCOV

2640

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

×

UNCOV

2642

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

×

UNCOV

2644

STableListInfo* tableListCreate() {

×

UNCOV

2645

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

×

UNCOV

2646

  if (pListInfo == NULL) {

×

2647

    return NULL;

×

UNCOV

2650

  pListInfo->remainGroups = NULL;

×

UNCOV

2651

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

×

UNCOV

2652

  if (pListInfo->pTableList == NULL) {

×

2653

    goto _error;

×

UNCOV

2656

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

×

UNCOV

2657

  if (pListInfo->map == NULL) {

×

2658

    goto _error;

×

UNCOV

2661

  pListInfo->numOfOuputGroups = 1;

×

UNCOV

2662

  return pListInfo;

×

2664

_error:

×

2665

  tableListDestroy(pListInfo);

×

2666

  return NULL;

×

UNCOV

2669

void tableListDestroy(STableListInfo* pTableListInfo) {

×

UNCOV

2670

  if (pTableListInfo == NULL) {

×

UNCOV

2671

    return;

×

UNCOV

2674

  taosArrayDestroy(pTableListInfo->pTableList);

×

UNCOV

2675

  taosMemoryFreeClear(pTableListInfo->groupOffset);

×

UNCOV

2677

  taosHashCleanup(pTableListInfo->map);

×

UNCOV

2678

  taosHashCleanup(pTableListInfo->remainGroups);

×

UNCOV

2679

  pTableListInfo->pTableList = NULL;

×

UNCOV

2680

  pTableListInfo->map = NULL;

×

UNCOV

2681

  taosMemoryFree(pTableListInfo);

×

UNCOV

2684

void tableListClear(STableListInfo* pTableListInfo) {

×

UNCOV

2685

  if (pTableListInfo == NULL) {

×

2686

    return;

×

UNCOV

2689

  taosArrayClear(pTableListInfo->pTableList);

×

UNCOV

2690

  taosHashClear(pTableListInfo->map);

×

UNCOV

2691

  taosHashClear(pTableListInfo->remainGroups);

×

UNCOV

2692

  taosMemoryFree(pTableListInfo->groupOffset);

×

UNCOV

2693

  pTableListInfo->numOfOuputGroups = 1;

×

UNCOV

2694

  pTableListInfo->oneTableForEachGroup = false;

×

UNCOV

2697

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

×

UNCOV

2698

  STableKeyInfo* pInfo1 = (STableKeyInfo*)p1;

×

UNCOV

2699

  STableKeyInfo* pInfo2 = (STableKeyInfo*)p2;

×

UNCOV

2701

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

×

UNCOV

2702

    return 0;

×

UNCOV

2704

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

×

UNCOV

2708

static int32_t sortTableGroup(STableListInfo* pTableListInfo) {

×

UNCOV

2709

  taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);

×

UNCOV

2710

  int32_t size = taosArrayGetSize(pTableListInfo->pTableList);

×

UNCOV

2712

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

×

UNCOV

2713

  if (!pList) {

×

2714

    return terrno;

×

UNCOV

2717

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

×

UNCOV

2718

  if (!pInfo) {

×

2719

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

×

2720

    return terrno;

×

UNCOV

2722

  uint64_t       gid = pInfo->groupId;

×

UNCOV

2724

  int32_t start = 0;

×

UNCOV

2725

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

×

UNCOV

2726

  if (!tmp) {

×

2727

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

×

2728

    return terrno;

×

UNCOV

2731

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

×

UNCOV

2732

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

×

UNCOV

2733

    if (!pInfo) {

×

UNCOV

2734

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

×

UNCOV

2735

      return terrno;

×

UNCOV

2737

    if (pInfo->groupId != gid) {

×

UNCOV

2738

      tmp = taosArrayPush(pList, &i);

×

UNCOV

2739

      if (!tmp) {

×

2740

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

×

2741

        return terrno;

×

UNCOV

2743

      gid = pInfo->groupId;

×

UNCOV

2747

  pTableListInfo->numOfOuputGroups = taosArrayGetSize(pList);

×

UNCOV

2748

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

×

UNCOV

2749

  if (pTableListInfo->groupOffset == NULL) {

×

2750

    taosArrayDestroy(pList);

×

2751

    return terrno;

×

UNCOV

2754

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

×

UNCOV

2755

  taosArrayDestroy(pList);

×

UNCOV

2756

  return TSDB_CODE_SUCCESS;

×

UNCOV

2759

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

×

UNCOV

2761

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

2763

  bool   groupByTbname = groupbyTbname(group);

×

UNCOV

2764

  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);

×

UNCOV

2765

  if (!numOfTables) {

×

2766

    return code;

×

UNCOV

2768

  if (group == NULL || groupByTbname) {

×

UNCOV

2769

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

×

UNCOV

2770

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

×

UNCOV

2771

      pTableListInfo->remainGroups =

×

UNCOV

2772

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

×

UNCOV

2773

      if (pTableListInfo->remainGroups == NULL) {

×

2774

        return terrno;

×

UNCOV

2777

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

×

UNCOV

2778

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

×

UNCOV

2779

        if (!info) {

×

2780

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

×

2781

          return terrno;

×

UNCOV

2783

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

×

UNCOV

2785

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

×

UNCOV

2786

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

×

UNCOV

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;

×

UNCOV

2793

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

×

UNCOV

2794

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

×

UNCOV

2795

        if (!info) {

×

2796

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

×

2797

          return terrno;

×

UNCOV

2799

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

×

UNCOV

2803

    pTableListInfo->oneTableForEachGroup = groupByTbname;

×

UNCOV

2804

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

×

UNCOV

2805

      pTableListInfo->oneTableForEachGroup = true;

×

UNCOV

2808

    if (groupSort && groupByTbname) {

×

UNCOV

2809

      taosArraySort(pTableListInfo->pTableList, orderbyGroupIdComparFn);

×

UNCOV

2810

      pTableListInfo->numOfOuputGroups = numOfTables;

×

UNCOV

2811

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

×

UNCOV

2812

      pTableListInfo->numOfOuputGroups = numOfTables;

×

UNCOV

2814

      pTableListInfo->numOfOuputGroups = 1;

×

UNCOV

2817

    bool initRemainGroups = false;

×

UNCOV

2818

    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pScanNode)) {

×

UNCOV

2819

      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pScanNode;

×

UNCOV

2820

      if (tsCountAlwaysReturnValue && pTableScanNode->needCountEmptyTable &&

×

UNCOV

2821

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

×

UNCOV

2822

        initRemainGroups = true;

×

UNCOV

2826

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

×

UNCOV

2827

    if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

2828

      return code;

×

UNCOV

2831

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

×

UNCOV

2833

    if (groupSort || pScanNode->groupOrderScan) {

×

UNCOV

2834

      code = sortTableGroup(pTableListInfo);

×

UNCOV

2839

  size_t size = taosArrayGetSize(pTableListInfo->pTableList);

×

UNCOV

2840

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

×

UNCOV

2841

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

×

UNCOV

2842

    if (!p) {

×

2843

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

×

2844

      return terrno;

×

UNCOV

2846

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

×

UNCOV

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;

×

UNCOV

2853

  return code;

×

UNCOV

2856

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

×

UNCOV

2859

  int64_t     st = taosGetTimestampUs();

×

UNCOV

2860

  const char* idStr = GET_TASKID(pTaskInfo);

×

UNCOV

2862

  if (pHandle == NULL) {

×

2863

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

×

2864

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

2867

  uint8_t digest[17] = {0};

×

UNCOV

2868

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

×

UNCOV

2870

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

2871

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

×

UNCOV

2872

    return code;

×

UNCOV

2875

  int32_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);

×

UNCOV

2877

  int64_t st1 = taosGetTimestampUs();

×

UNCOV

2878

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

×

UNCOV

2879

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

×

UNCOV

2882

  if (numOfTables == 0) {

×

UNCOV

2883

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

×

UNCOV

2884

    return TSDB_CODE_SUCCESS;

×

UNCOV

2887

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

×

UNCOV

2889

  if (code != TSDB_CODE_SUCCESS) {

×

UNCOV

2890

    return code;

×

UNCOV

2893

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

×

UNCOV

2894

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

×

UNCOV

2896

  return TSDB_CODE_SUCCESS;

×

UNCOV

2899

char* getStreamOpName(uint16_t opType) {

×

UNCOV

2900

  switch (opType) {

×

UNCOV

2901

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:

×

UNCOV

2902

      return "stream scan";

×

UNCOV

2903

    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:

×

UNCOV

2904

      return "project";

×

UNCOV

2905

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:

×

UNCOV

2906

      return "interval single";

×

UNCOV

2907

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL:

×

UNCOV

2908

      return "interval final";

×

UNCOV

2909

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL:

×

UNCOV

2910

      return "interval semi";

×

UNCOV

2911

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL:

×

UNCOV

2912

      return "interval mid";

×

UNCOV

2913

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL:

×

UNCOV

2914

      return "stream fill";

×

UNCOV

2915

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION:

×

UNCOV

2916

      return "session single";

×

UNCOV

2917

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION:

×

UNCOV

2918

      return "session semi";

×

UNCOV

2919

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION:

×

UNCOV

2920

      return "session final";

×

UNCOV

2921

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE:

×

UNCOV

2922

      return "state single";

×

UNCOV

2923

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION:

×

UNCOV

2924

      return "stream partitionby";

×

UNCOV

2925

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT:

×

UNCOV

2926

      return "stream event";

×

UNCOV

2927

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT:

×

UNCOV

2928

      return "stream count";

×

UNCOV

2929

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC:

×

UNCOV

2930

      return "stream interp";

×

UNCOV

2931

    case QUERY_NODE_PHYSICAL_PLAN_STREAM_CONTINUE_INTERVAL:

×

UNCOV

2932

      return "interval continue";

×

2934

  return "";

×

UNCOV

2937

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

×

UNCOV

2938

  if (!pBlock) {

×

2939

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

×

2940

    return;

×

UNCOV

2941

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

×

UNCOV

2942

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

×

UNCOV

2943

    return;

×

UNCOV

2945

  if (qDebugFlag & DEBUG_DEBUG) {

×

UNCOV

2946

    char*   pBuf = NULL;

×

UNCOV

2947

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

×

UNCOV

2948

    if (code == 0) {

×

UNCOV

2949

      qDebug("%s", pBuf);

×

UNCOV

2950

      taosMemoryFree(pBuf);

×

UNCOV

2955

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

×

UNCOV

2956

  if (!pBlock) {

×

2957

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

×

2958

    return;

×

UNCOV

2959

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

×

UNCOV

2960

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

×

UNCOV

2963

    return;

×

UNCOV

2965

  if (qDebugFlag & DEBUG_DEBUG) {

×

UNCOV

2966

    char* pBuf = NULL;

×

UNCOV

2968

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

×

UNCOV

2969

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

×

UNCOV

2970

    if (code == 0) {

×

UNCOV

2971

      qDebug("%s", pBuf);

×

UNCOV

2972

      taosMemoryFree(pBuf);

×

UNCOV

2977

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

×

UNCOV

2979

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

×

UNCOV

2980

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

×

UNCOV

2982

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

×

UNCOV

2983

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

×

UNCOV

2984

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

×

UNCOV

2985

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

×

UNCOV

2986

×

UNCOV

2988

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

×

UNCOV

2990

  SColumnDataAgg* pColAgg = NULL;

×

UNCOV

2991

  const char*     isNull = oldkeyBuf;

×

UNCOV

2992

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

×

UNCOV

2994

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

×

UNCOV

2995

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

×

UNCOV

2996

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

×

UNCOV

2997

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

×

UNCOV

2999

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

×

UNCOV

3000

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

×

UNCOV

3002

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

×

UNCOV

3003

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

×

UNCOV

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;

×

UNCOV

3008

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

×

UNCOV

3009

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

×

UNCOV

3010

        p += varDataTLen(val);

×

UNCOV

3012

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

×

UNCOV

3013

        p += pCol->bytes;

×

UNCOV

3017

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

×

UNCOV

3018

  return 0;

×

UNCOV

3021

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

×

UNCOV

3022

  uint32_t        colNum = pSortGroupCols->size;

×

UNCOV

3023

  SColumnDataAgg* pColAgg = NULL;

×

UNCOV

3024

  char*           isNull = keyBuf;

×

UNCOV

3025

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

×

UNCOV

3027

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

×

UNCOV

3028

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

×

UNCOV

3029

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

×

UNCOV

3030

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

×

UNCOV

3032

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

×

UNCOV

3034

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

×

UNCOV

3035

      isNull[i] = 1;

×

UNCOV

3037

      isNull[i] = 0;

×

UNCOV

3038

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

×

UNCOV

3039

      if (pCol->type == TSDB_DATA_TYPE_JSON) {

×

3040

        int32_t len = getJsonValueLen(val);

×

3041

        memcpy(p, val, len);

×

3042

        p += len;

×

UNCOV

3043

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

×

UNCOV

3044

        varDataCopy(p, val);

×

UNCOV

3045

        p += varDataTLen(val);

×

UNCOV

3047

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

×

UNCOV

3048

        p += pCol->bytes;

×

UNCOV

3052

  return (int32_t)(p - keyBuf);

×

UNCOV

3055

uint64_t calcGroupId(char* pData, int32_t len) {

×

UNCOV

3057

  tMD5Init(&context);

×

UNCOV

3058

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

×

UNCOV

3059

  tMD5Final(&context);

×

UNCOV

3062

  uint64_t id = 0;

×

UNCOV

3063

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

×

UNCOV

3064

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

×

UNCOV

3065

  return id;

×

UNCOV

3068

SNodeList* makeColsNodeArrFromSortKeys(SNodeList* pSortKeys) {

×

UNCOV

3070

  SNodeList* ret = NULL;

×

UNCOV

3071

  FOREACH(node, pSortKeys) {

×

UNCOV

3072

    SOrderByExprNode* pSortKey = (SOrderByExprNode*)node;

×

UNCOV

3073

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

×

UNCOV

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;

×

UNCOV

3080

  return ret;

×

UNCOV

3083

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

×

UNCOV

3084

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

3085

  int32_t lino = 0;

×

UNCOV

3086

  int32_t len = 0;

×

UNCOV

3087

  int32_t keyNum = taosArrayGetSize(keys);

×

UNCOV

3088

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

×

UNCOV

3089

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

×

UNCOV

3090

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

×

UNCOV

3091

    len += pCol->bytes;

×

UNCOV

3093

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

×

UNCOV

3094

  *pLen = len;

×

UNCOV

3096

_end:

×

UNCOV

3097

  if (code != TSDB_CODE_SUCCESS) {

×

3098

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

×

UNCOV

3100

  return code;

×

taosdata / TDengine / #3621

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

taosdata / TDengine / #3621

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