taosdata/TDengine | Build 4986 | source/libs/executor/src/dynqueryctrloperator.c | Coveralls

36

void freeVgTableList(void* ptr) {

×

37

  taosArrayDestroy(*(SArray**)ptr);

×

38

×

40

static void destroyStbJoinTableList(SStbJoinTableList* pListHead) {

×

41

  SStbJoinTableList* pNext = NULL;

×

43

  while (pListHead) {

×

44

    taosMemoryFree(pListHead->pLeftVg);

×

45

    taosMemoryFree(pListHead->pLeftUid);

×

46

    taosMemoryFree(pListHead->pRightVg);

×

47

    taosMemoryFree(pListHead->pRightUid);

×

48

    pNext = pListHead->pNext;

×

49

    taosMemoryFree(pListHead);

×

50

    pListHead = pNext;

×

52

×

54

static void destroyStbJoinDynCtrlInfo(SStbJoinDynCtrlInfo* pStbJoin) {

×

55

  qDebug("dynQueryCtrl exec info, prevBlk:%" PRId64 ", prevRows:%" PRId64 ", postBlk:%" PRId64 ", postRows:%" PRId64 ", leftCacheNum:%" PRId64 ", rightCacheNum:%" PRId64,

×

59

  if (pStbJoin->basic.batchFetch) {

×

60

    if (pStbJoin->ctx.prev.leftHash) {

×

61

      tSimpleHashSetFreeFp(pStbJoin->ctx.prev.leftHash, freeVgTableList);

×

62

      tSimpleHashCleanup(pStbJoin->ctx.prev.leftHash);

×

64

    if (pStbJoin->ctx.prev.rightHash) {

×

65

      tSimpleHashSetFreeFp(pStbJoin->ctx.prev.rightHash, freeVgTableList);

×

66

      tSimpleHashCleanup(pStbJoin->ctx.prev.rightHash);

×

69

    if (pStbJoin->ctx.prev.leftCache) {

×

70

      tSimpleHashCleanup(pStbJoin->ctx.prev.leftCache);

×

72

    if (pStbJoin->ctx.prev.rightCache) {

×

73

      tSimpleHashCleanup(pStbJoin->ctx.prev.rightCache);

×

75

    if (pStbJoin->ctx.prev.onceTable) {

×

76

      tSimpleHashCleanup(pStbJoin->ctx.prev.onceTable);

×

80

  destroyStbJoinTableList(pStbJoin->ctx.prev.pListHead);

×

81

×

83

void destroyColRefInfo(void *info) {

×

84

  SColRefInfo *pColRefInfo = (SColRefInfo *)info;

×

85

  if (pColRefInfo) {

×

86

    taosMemoryFree(pColRefInfo->colName);

×

87

    taosMemoryFree(pColRefInfo->colrefName);

×

89

×

91

void destroyColRefArray(void *info) {

×

92

  SArray *pColRefArray = *(SArray **)info;

×

93

  if (pColRefArray) {

×

94

    taosArrayDestroyEx(pColRefArray, destroyColRefInfo);

×

96

×

98

void freeUseDbOutput(void* pOutput) {

×

99

  SUseDbOutput *pOut = *(SUseDbOutput**)pOutput;

×

100

  if (NULL == pOutput) {

×

101

    return;

×

104

  if (pOut->dbVgroup) {

×

105

    freeVgInfo(pOut->dbVgroup);

×

107

  taosMemFree(pOut);

×

110

void destroyOtbInfoArray(void *info) {

×

111

  SArray *pOtbInfoArray = *(SArray **)info;

×

112

  if (pOtbInfoArray) {

×

113

    taosArrayDestroyEx(pOtbInfoArray, destroySOrgTbInfo);

×

115

×

117

void destroyOtbVgIdToOtbInfoArrayMap(void *info) {

×

118

  SHashObj* pOtbVgIdToOtbInfoArrayMap = *(SHashObj **)info;

×

119

  if (pOtbVgIdToOtbInfoArrayMap) {

×

120

    taosHashSetFreeFp(pOtbVgIdToOtbInfoArrayMap, destroyOtbInfoArray);

×

121

    taosHashCleanup(pOtbVgIdToOtbInfoArrayMap);

×

123

×

125

void destroyTagList(void *info) {

×

126

  SArray *pTagList = *(SArray **)info;

×

127

  if (pTagList) {

×

128

    taosArrayDestroyEx(pTagList, destroyTagVal);

×

130

×

132

static void destroyRefColIdGroup(void *info) {

×

133

  SRefColIdGroup *pGroup = (SRefColIdGroup *)info;

×

134

  if (pGroup && pGroup->pSlotIdList) {

×

135

    taosArrayDestroy(pGroup->pSlotIdList);

×

136

    pGroup->pSlotIdList = NULL;

×

138

×

140

void destroyVtbUidTagListMap(void *info) {

×

141

  SHashObj* pVtbUidTagListMap = *(SHashObj **)info;

×

142

  if (pVtbUidTagListMap) {

×

143

    taosHashSetFreeFp(pVtbUidTagListMap, destroyTagList);

×

144

    taosHashCleanup(pVtbUidTagListMap);

×

146

×

148

static void destroyVtbScanDynCtrlInfo(SVtbScanDynCtrlInfo* pVtbScan) {

×

149

  if (pVtbScan->dbName) {

×

150

    taosMemoryFreeClear(pVtbScan->dbName);

×

152

  if (pVtbScan->tbName) {

×

153

    taosMemoryFreeClear(pVtbScan->tbName);

×

155

  if (pVtbScan->refColGroups) {

×

156

    taosArrayDestroyEx(pVtbScan->refColGroups, destroyRefColIdGroup);

×

157

    pVtbScan->refColGroups = NULL;

×

159

  if (pVtbScan->childTableList) {

×

160

    taosArrayDestroyEx(pVtbScan->childTableList, destroyColRefArray);

×

162

  if (pVtbScan->colRefInfo) {

×

163

    taosArrayDestroyEx(pVtbScan->colRefInfo, destroyColRefInfo);

×

164

    pVtbScan->colRefInfo = NULL;

×

166

  if (pVtbScan->childTableMap) {

×

167

    taosHashCleanup(pVtbScan->childTableMap);

×

169

  if (pVtbScan->readColList) {

×

170

    taosArrayDestroy(pVtbScan->readColList);

×

172

  if (pVtbScan->readColSet) {

×

173

    taosHashCleanup(pVtbScan->readColSet);

×

175

  if (pVtbScan->dbVgInfoMap) {

×

176

    taosHashSetFreeFp(pVtbScan->dbVgInfoMap, freeUseDbOutput);

×

177

    taosHashCleanup(pVtbScan->dbVgInfoMap);

×

179

  if (pVtbScan->otbNameToOtbInfoMap) {

×

180

    taosHashSetFreeFp(pVtbScan->otbNameToOtbInfoMap, destroySOrgTbInfo);

×

181

    taosHashCleanup(pVtbScan->otbNameToOtbInfoMap);

×

183

  if (pVtbScan->pRsp) {

×

184

    tFreeSUsedbRsp(pVtbScan->pRsp);

×

185

    taosMemoryFreeClear(pVtbScan->pRsp);

×

187

  if (pVtbScan->existOrgTbVg) {

×

188

    taosHashCleanup(pVtbScan->existOrgTbVg);

×

190

  if (pVtbScan->curOrgTbVg) {

×

191

    taosHashCleanup(pVtbScan->curOrgTbVg);

×

193

  if (pVtbScan->newAddedVgInfo) {

×

194

    taosHashCleanup(pVtbScan->newAddedVgInfo);

×

196

  if (pVtbScan->otbVgIdToOtbInfoArrayMap) {

×

197

    taosHashSetFreeFp(pVtbScan->otbVgIdToOtbInfoArrayMap, destroyOtbInfoArray);

×

198

    taosHashCleanup(pVtbScan->otbVgIdToOtbInfoArrayMap);

×

200

  if (pVtbScan->vtbUidToVgIdMapMap) {

×

201

    taosHashSetFreeFp(pVtbScan->vtbUidToVgIdMapMap, destroyOtbVgIdToOtbInfoArrayMap);

×

202

    taosHashCleanup(pVtbScan->vtbUidToVgIdMapMap);

×

204

  if (pVtbScan->vtbGroupIdToVgIdMapMap) {

×

205

    taosHashSetFreeFp(pVtbScan->vtbGroupIdToVgIdMapMap, destroyOtbVgIdToOtbInfoArrayMap);

×

206

    taosHashCleanup(pVtbScan->vtbGroupIdToVgIdMapMap);

×

208

  if (pVtbScan->vtbUidTagListMap) {

×

209

    taosHashSetFreeFp(pVtbScan->vtbUidTagListMap, destroyTagList);

×

210

    taosHashCleanup(pVtbScan->vtbUidTagListMap);

×

212

  if (pVtbScan->vtbGroupIdTagListMap) {

×

213

    taosHashCleanup(pVtbScan->vtbGroupIdTagListMap);

×

215

  if (pVtbScan->vtbUidToGroupIdMap) {

×

216

    taosHashCleanup(pVtbScan->vtbUidToGroupIdMap);

×

218

×

220

void destroyWinArray(void *info) {

×

221

  SArray *pWinArray = *(SArray **)info;

×

222

  if (pWinArray) {

×

223

    taosArrayDestroy(pWinArray);

×

225

×

227

static void destroyVtbWindowDynCtrlInfo(SVtbWindowDynCtrlInfo* pVtbWindow) {

×

228

  if (pVtbWindow->pRes) {

×

229

    blockDataDestroy(pVtbWindow->pRes);

×

231

  if (pVtbWindow->pWins) {

×

232

    taosArrayDestroyEx(pVtbWindow->pWins, destroyWinArray);

×

234

×

236

static void destroyDynQueryCtrlOperator(void* param) {

×

237

  SDynQueryCtrlOperatorInfo* pDyn = (SDynQueryCtrlOperatorInfo*)param;

×

239

  switch (pDyn->qType) {

×

240

    case DYN_QTYPE_STB_HASH:

×

241

      destroyStbJoinDynCtrlInfo(&pDyn->stbJoin);

×

242

      break;

×

243

    case DYN_QTYPE_VTB_WINDOW:

×

244

      destroyVtbWindowDynCtrlInfo(&pDyn->vtbWindow);

×

245

      destroyVtbScanDynCtrlInfo(&pDyn->vtbScan);

×

246

      break;

×

247

    case DYN_QTYPE_VTB_INTERVAL:

×

251

      destroyVtbScanDynCtrlInfo(&pDyn->vtbScan);

×

252

      break;

×

253

    default:

×

254

      qError("unsupported dynamic query ctrl type: %d", pDyn->qType);

×

255

      break;

×

258

  taosMemoryFreeClear(param);

×

259

×

262

  if (batchFetch) {

×

263

    return true;

×

266

  if (rightTable) {

×

267

    return pPost->rightCurrUid == pPost->rightNextUid;

×

270

  uint32_t* num = tSimpleHashGet(pPrev->leftCache, &uid, sizeof(uid));

×

272

  return (NULL == num) ? false : true;

×

275

static int32_t updatePostJoinCurrTableInfo(SStbJoinDynCtrlInfo*          pStbJoin) {

×

276

  SStbJoinPrevJoinCtx*       pPrev = &pStbJoin->ctx.prev;

×

277

  SStbJoinPostJoinCtx*       pPost = &pStbJoin->ctx.post;

×

278

  SStbJoinTableList*         pNode = pPrev->pListHead;

×

279

  int32_t*                   leftVgId = pNode->pLeftVg + pNode->readIdx;

×

280

  int32_t*                   rightVgId = pNode->pRightVg + pNode->readIdx;

×

281

  int64_t*                   leftUid = pNode->pLeftUid + pNode->readIdx;

×

282

  int64_t*                   rightUid = pNode->pRightUid + pNode->readIdx;

×

283

  int64_t                    readIdx = pNode->readIdx + 1;

×

284

  int64_t                    rightPrevUid = pPost->rightCurrUid;

×

286

  pPost->leftCurrUid = *leftUid;

×

287

  pPost->rightCurrUid = *rightUid;

×

289

  pPost->leftVgId = *leftVgId;

×

290

  pPost->rightVgId = *rightVgId;

×

293

    if (readIdx < pNode->uidNum) {

×

294

      pPost->rightNextUid = *(pNode->pRightUid + readIdx);

×

295

      break;

×

298

    pNode = pNode->pNext;

×

299

    if (NULL == pNode) {

×

300

      pPost->rightNextUid = 0;

×

301

      break;

×

304

    rightUid = pNode->pRightUid;

×

305

    readIdx = 0;

×

308

  pPost->leftNeedCache = tableNeedCache(*leftUid, pPrev, pPost, false, pStbJoin->basic.batchFetch);

×

309

  pPost->rightNeedCache = tableNeedCache(*rightUid, pPrev, pPost, true, pStbJoin->basic.batchFetch);

×

311

  if (!pStbJoin->basic.batchFetch && pPost->rightNeedCache && rightPrevUid != pPost->rightCurrUid) {

×

312

    QRY_ERR_RET(tSimpleHashPut(pPrev->rightCache, &pPost->rightCurrUid, sizeof(pPost->rightCurrUid), NULL, 0));

×

313

    pStbJoin->execInfo.rightCacheNum++;

×

316

  return TSDB_CODE_SUCCESS;

×

319

static int32_t copyOrgTbInfo(SOrgTbInfo* pSrc, SOrgTbInfo** ppDst) {

×

320

  int32_t     code = TSDB_CODE_SUCCESS;

×

321

  int32_t     lino = 0;

×

322

  SOrgTbInfo* pTbInfo = NULL;

×

324

  qDebug("start to copy org table info, vgId:%d, tbName:%s", pSrc->vgId, pSrc->tbName);

×

326

  pTbInfo = taosMemoryMalloc(sizeof(SOrgTbInfo));

×

327

  QUERY_CHECK_NULL(pTbInfo, code, lino, _return, terrno)

×

329

  pTbInfo->vgId = pSrc->vgId;

×

330

  tstrncpy(pTbInfo->tbName, pSrc->tbName, TSDB_TABLE_FNAME_LEN);

×

332

  pTbInfo->colMap = taosArrayDup(pSrc->colMap, NULL);

×

333

  QUERY_CHECK_NULL(pTbInfo->colMap, code, lino, _return, terrno)

×

335

  *ppDst = pTbInfo;

×

337

  return code;

×

338

_return:

×

339

  qError("failed to copy org table info, code:%d, line:%d", code, lino);

×

340

  if (pTbInfo) {

×

341

    if (pTbInfo->colMap) {

×

342

      taosArrayDestroy(pTbInfo->colMap);

×

344

    taosMemoryFreeClear(pTbInfo);

×

346

  return code;

×

349

static int32_t buildTagListForExchangeBasicParam(SExchangeOperatorBasicParam* pBasic, SArray* pTagList) {

×

350

  int32_t  code = TSDB_CODE_SUCCESS;

×

351

  int32_t  lino = 0;

×

354

  pBasic->tagList = taosArrayInit(1, sizeof(STagVal));

×

355

  QUERY_CHECK_NULL(pBasic->tagList, code, lino, _return, terrno)

×

357

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

×

358

    STagVal* pSrcTag = (STagVal*)taosArrayGet(pTagList, i);

×

359

    QUERY_CHECK_NULL(pSrcTag, code, lino, _return, terrno)

×

360

    tmpTag.type = pSrcTag->type;

×

361

    tmpTag.cid = pSrcTag->cid;

×

362

    if (IS_VAR_DATA_TYPE(pSrcTag->type)) {

×

363

      tmpTag.nData = pSrcTag->nData;

×

364

      tmpTag.pData = taosMemoryMalloc(tmpTag.nData);

×

365

      QUERY_CHECK_NULL(tmpTag.pData, code, lino, _return, terrno)

×

366

      memcpy(tmpTag.pData, pSrcTag->pData, tmpTag.nData);

×

368

      tmpTag.i64 = pSrcTag->i64;

×

371

    QUERY_CHECK_NULL(taosArrayPush(pBasic->tagList, &tmpTag), code, lino, _return, terrno)

×

372

    tmpTag = (STagVal){0};

×

375

  return code;

×

376

_return:

×

377

  if (pBasic->tagList) {

×

378

    taosArrayDestroyEx(pBasic->tagList, destroyTagVal);

×

379

    pBasic->tagList = NULL;

×

381

  if (tmpTag.pData) {

×

382

    taosMemoryFree(tmpTag.pData);

×

384

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

×

385

  return code;

×

388

static int32_t buildBatchOrgTbInfoForExchangeBasicParam(SExchangeOperatorBasicParam* pBasic, SArray* pOrgTbInfoArray) {

×

389

  int32_t     code = TSDB_CODE_SUCCESS;

×

390

  int32_t     lino = 0;

×

393

  pBasic->batchOrgTbInfo = taosArrayInit(1, sizeof(SOrgTbInfo));

×

394

  QUERY_CHECK_NULL(pBasic->batchOrgTbInfo, code, lino, _return, terrno)

×

396

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

×

397

    SOrgTbInfo* pSrc = (SOrgTbInfo*)taosArrayGet(pOrgTbInfoArray, i);

×

398

    QUERY_CHECK_NULL(pSrc, code, lino, _return, terrno)

×

399

    batchInfo.vgId = pSrc->vgId;

×

400

    tstrncpy(batchInfo.tbName, pSrc->tbName, TSDB_TABLE_FNAME_LEN);

×

401

    batchInfo.colMap = taosArrayDup(pSrc->colMap, NULL);

×

402

    QUERY_CHECK_NULL(batchInfo.colMap, code, lino, _return, terrno)

×

403

    QUERY_CHECK_NULL(taosArrayPush(pBasic->batchOrgTbInfo, &batchInfo), code, lino, _return, terrno)

×

404

    batchInfo = (SOrgTbInfo){0};

×

407

  return code;

×

408

_return:

×

409

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

×

410

  if (pBasic->batchOrgTbInfo) {

×

411

    taosArrayDestroyEx(pBasic->batchOrgTbInfo, destroySOrgTbInfo);

×

412

    pBasic->batchOrgTbInfo = NULL;

×

414

  if (batchInfo.colMap) {

×

415

    taosArrayDestroy(batchInfo.colMap);

×

416

    batchInfo.colMap = NULL;

×

418

  return code;

×

421

static int32_t buildGroupCacheOperatorParam(SOperatorParam** ppRes, int32_t downstreamIdx, int32_t vgId, int64_t tbUid, bool needCache, SOperatorParam* pChild) {

×

422

  int32_t code = TSDB_CODE_SUCCESS;

×

423

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

424

  if (NULL == *ppRes) {

×

425

    code = terrno;

×

426

    freeOperatorParam(pChild, OP_GET_PARAM);

×

427

    return code;

×

429

  if (pChild) {

×

430

    (*ppRes)->pChildren = taosArrayInit(1, POINTER_BYTES);

×

431

    if (NULL == (*ppRes)->pChildren) {

×

432

      code = terrno;

×

433

      freeOperatorParam(pChild, OP_GET_PARAM);

×

434

      freeOperatorParam(*ppRes, OP_GET_PARAM);

×

435

      *ppRes = NULL;

×

436

      return code;

×

438

    if (NULL == taosArrayPush((*ppRes)->pChildren, &pChild)) {

×

439

      code = terrno;

×

440

      freeOperatorParam(pChild, OP_GET_PARAM);

×

441

      freeOperatorParam(*ppRes, OP_GET_PARAM);

×

442

      *ppRes = NULL;

×

443

      return code;

×

446

    (*ppRes)->pChildren = NULL;

×

449

  SGcOperatorParam* pGc = taosMemoryMalloc(sizeof(SGcOperatorParam));

×

450

  if (NULL == pGc) {

×

451

    code = terrno;

×

452

    freeOperatorParam(*ppRes, OP_GET_PARAM);

×

453

    *ppRes = NULL;

×

454

    return code;

×

457

  pGc->sessionId = atomic_add_fetch_64(&gSessionId, 1);

×

458

  pGc->downstreamIdx = downstreamIdx;

×

459

  pGc->vgId = vgId;

×

460

  pGc->tbUid = tbUid;

×

461

  pGc->needCache = needCache;

×

463

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE;

×

464

  (*ppRes)->downstreamIdx = downstreamIdx;

×

465

  (*ppRes)->value = pGc;

×

466

  (*ppRes)->reUse = false;

×

468

  return TSDB_CODE_SUCCESS;

×

472

static int32_t buildGroupCacheNotifyOperatorParam(SOperatorParam** ppRes, int32_t downstreamIdx, int32_t vgId, int64_t tbUid) {

×

473

  int32_t code = TSDB_CODE_SUCCESS;

×

474

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

475

  if (NULL == *ppRes) {

×

476

    return terrno;

×

478

  (*ppRes)->pChildren = NULL;

×

480

  SGcNotifyOperatorParam* pGc = taosMemoryMalloc(sizeof(SGcNotifyOperatorParam));

×

481

  if (NULL == pGc) {

×

482

    code = terrno;

×

483

    freeOperatorParam(*ppRes, OP_NOTIFY_PARAM);

×

484

    return code;

×

487

  pGc->downstreamIdx = downstreamIdx;

×

488

  pGc->vgId = vgId;

×

489

  pGc->tbUid = tbUid;

×

491

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_GROUP_CACHE;

×

492

  (*ppRes)->downstreamIdx = downstreamIdx;

×

493

  (*ppRes)->value = pGc;

×

494

  (*ppRes)->reUse = false;

×

496

  return TSDB_CODE_SUCCESS;

×

499

static int32_t buildExchangeOperatorBasicParam(SExchangeOperatorBasicParam* pBasic, ENodeType srcOpType,

×

505

  int32_t code = TSDB_CODE_SUCCESS;

×

506

  int32_t lino = 0;

×

508

  qDebug("buildExchangeOperatorBasicParam, srcOpType:%d, exchangeType:%d, vgId:%d, groupId:%" PRIu64 ", tableSeq:%d, "

×

511

  pBasic->paramType = DYN_TYPE_EXCHANGE_PARAM;

×

512

  pBasic->srcOpType = srcOpType;

×

513

  pBasic->vgId = vgId;

×

514

  pBasic->groupid = groupId;

×

515

  pBasic->window = window;

×

516

  pBasic->tableSeq = tableSeq;

×

517

  pBasic->type = exchangeType;

×

518

  pBasic->isNewParam = isNewParam;

×

520

  if (pDownstreamSourceNode) {

×

521

    pBasic->isNewDeployed = true;

×

522

    pBasic->newDeployedSrc.type = QUERY_NODE_DOWNSTREAM_SOURCE;

×

523

    pBasic->newDeployedSrc.clientId = pDownstreamSourceNode->clientId;// current task's taskid

×

524

    pBasic->newDeployedSrc.taskId = pDownstreamSourceNode->taskId;

×

525

    pBasic->newDeployedSrc.fetchMsgType = TDMT_STREAM_FETCH;

×

526

    pBasic->newDeployedSrc.localExec = false;

×

527

    pBasic->newDeployedSrc.addr.nodeId = pDownstreamSourceNode->addr.nodeId;

×

528

    memcpy(&pBasic->newDeployedSrc.addr.epSet, &pDownstreamSourceNode->addr.epSet, sizeof(SEpSet));

×

530

    pBasic->isNewDeployed = false;

×

531

    pBasic->newDeployedSrc = (SDownstreamSourceNode){0};

×

534

  if (pUidList) {

×

535

    pBasic->uidList = taosArrayDup(pUidList, NULL);

×

536

    QUERY_CHECK_NULL(pBasic->uidList, code, lino, _return, terrno)

×

538

    pBasic->uidList = taosArrayInit(1, sizeof(int64_t));

×

539

    QUERY_CHECK_NULL(pBasic->uidList, code, lino, _return, terrno)

×

542

  if (pOrgTbInfo) {

×

543

    code = copyOrgTbInfo(pOrgTbInfo, &pBasic->orgTbInfo);

×

544

    QUERY_CHECK_CODE(code, lino, _return);

×

546

    pBasic->orgTbInfo = NULL;

×

549

  if (pTagList) {

×

550

    code = buildTagListForExchangeBasicParam(pBasic, pTagList);

×

551

    QUERY_CHECK_CODE(code, lino, _return);

×

553

    pBasic->tagList = NULL;

×

556

  if (pOrgTbInfoArray) {

×

557

    code = buildBatchOrgTbInfoForExchangeBasicParam(pBasic, pOrgTbInfoArray);

×

558

    QUERY_CHECK_CODE(code, lino, _return);

×

560

    pBasic->batchOrgTbInfo = NULL;

×

562

  return code;

×

564

_return:

×

565

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

×

566

  freeExchangeGetBasicOperatorParam(pBasic);

×

567

  return code;

×

570

static int32_t buildExchangeOperatorParamImpl(SOperatorParam** ppRes, int32_t downstreamIdx, ENodeType srcOpType,

×

577

  int32_t                      code = TSDB_CODE_SUCCESS;

×

578

  int32_t                      lino = 0;

×

579

  SOperatorParam*              pParam = NULL;

×

580

  SExchangeOperatorParam*      pExc = NULL;

×

582

  *ppRes = NULL;

×

584

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

585

  QUERY_CHECK_NULL(pParam, code, lino, _return, terrno)

×

587

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;

×

588

  pParam->downstreamIdx = downstreamIdx;

×

589

  pParam->reUse = reUse;

×

590

  pParam->pChildren = NULL;

×

591

  pParam->value = taosMemoryMalloc(sizeof(SExchangeOperatorParam));

×

592

  QUERY_CHECK_NULL(pParam->value, code, lino, _return, terrno)

×

594

  pExc = (SExchangeOperatorParam*)pParam->value;

×

595

  pExc->multiParams = false;

×

597

  code = buildExchangeOperatorBasicParam(&pExc->basic, srcOpType, exchangeType, vgId, groupId,

×

600

  QUERY_CHECK_CODE(code, lino, _return);

×

602

  *ppRes = pParam;

×

603

  return code;

×

604

_return:

×

605

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

×

606

  if (pParam) {

×

607

    freeOperatorParam(pParam, OP_GET_PARAM);

×

609

  return code;

×

612

static int32_t buildExchangeOperatorParam(SOperatorParam** ppRes, int32_t downstreamIdx, const int32_t* pVgId, int64_t* pUid) {

×

613

  int32_t code = TSDB_CODE_SUCCESS;

×

614

  int32_t lino = 0;

×

616

  SArray* pUidList = taosArrayInit(1, sizeof(int64_t));

×

617

  QUERY_CHECK_NULL(pUidList, code, lino, _return, terrno)

×

619

  QUERY_CHECK_NULL(taosArrayPush(pUidList, pUid), code, lino, _return, terrno);

×

621

  code = buildExchangeOperatorParamImpl(ppRes, downstreamIdx, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, EX_SRC_TYPE_STB_JOIN_SCAN,

×

622

                                        *pVgId, 0, pUidList, NULL, NULL, NULL, (STimeWindow){0}, NULL, true, false, false, false);

×

623

  QUERY_CHECK_CODE(code, lino, _return);

×

625

_return:

×

626

  if (code) {

×

627

    qError("failed to build exchange operator param, code:%d", code);

×

629

  taosArrayDestroy(pUidList);

×

630

  return code;

×

633

static int32_t buildExchangeOperatorParamForExternalWindow(SOperatorParam** ppRes, int32_t downstreamIdx, STimeWindow win) {

×

634

  int32_t                   code = TSDB_CODE_SUCCESS;

×

635

  int32_t                   lino = 0;

×

637

  code = buildExchangeOperatorParamImpl(ppRes, downstreamIdx, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, EX_SRC_TYPE_VTB_WIN_SCAN,

×

639

  QUERY_CHECK_CODE(code, lino, _return);

×

641

  return code;

×

642

_return:

×

643

  qError("failed to build exchange operator param for external window, code:%d", code);

×

644

  return code;

×

647

static int32_t buildExchangeOperatorParamForVTagScan(SOperatorParam** ppRes, int32_t downstreamIdx, int32_t vgId, tb_uid_t uid) {

×

648

  int32_t                      code = TSDB_CODE_SUCCESS;

×

649

  int32_t                      lino = 0;

×

650

  SArray*                      pUidList = NULL;

×

652

  pUidList = taosArrayInit(1, sizeof(int64_t));

×

653

  QUERY_CHECK_NULL(pUidList, code, lino, _return, terrno)

×

655

  QUERY_CHECK_NULL(taosArrayPush(pUidList, &uid), code, lino, _return, terrno)

×

657

  code = buildExchangeOperatorParamImpl(ppRes, downstreamIdx, QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN, EX_SRC_TYPE_VSTB_TAG_SCAN,

×

658

                                        vgId, 0, pUidList, NULL, NULL, NULL, (STimeWindow){0}, NULL, false, false, true, false);

×

659

  QUERY_CHECK_CODE(code, lino, _return);

×

661

_return:

×

662

  if (code) {

×

663

    qError("failed to build exchange operator param for tag scan, code:%d", code);

×

665

  taosArrayDestroy(pUidList);

×

666

  return code;

×

669

static int32_t buildExchangeOperatorParamForVScan(SOperatorParam** ppRes, int32_t downstreamIdx, SOrgTbInfo* pOrgTbInfo,

×

671

  int32_t                      code = TSDB_CODE_SUCCESS;

×

672

  int32_t                      lino = 0;

×

674

  code = buildExchangeOperatorParamImpl(ppRes, downstreamIdx, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, EX_SRC_TYPE_VSTB_SCAN,

×

675

                                        pOrgTbInfo->vgId, 0, NULL, pOrgTbInfo, NULL, NULL, (STimeWindow){0}, pNewSource, false, true, true, true);

×

676

  QUERY_CHECK_CODE(code, lino, _return);

×

678

  return code;

×

679

_return:

×

680

  qError("failed to build exchange operator param for vscan, code:%d", code);

×

681

  return code;

×

684

static int32_t buildBatchExchangeOperatorParam(SOperatorParam** ppRes, int32_t downstreamIdx, SSHashObj* pVg) {

×

685

  int32_t                       code = TSDB_CODE_SUCCESS;

×

686

  int32_t                       line = 0;

×

687

  SOperatorParam*               pParam = NULL;

×

688

  SExchangeOperatorBatchParam*  pExc = NULL;

×

689

  SExchangeOperatorBasicParam   basic = {0};

×

691

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

692

  QUERY_CHECK_NULL(pParam, code, line, _return, terrno);

×

694

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;

×

695

  pParam->downstreamIdx = downstreamIdx;

×

696

  pParam->reUse = false;

×

697

  pParam->pChildren = NULL;

×

698

  pParam->value = taosMemoryMalloc(sizeof(SExchangeOperatorParam));

×

699

  QUERY_CHECK_NULL(pParam->value, code, line, _return, terrno);

×

701

  pExc = pParam->value;

×

702

  pExc->multiParams = true;

×

703

  pExc->pBatchs = tSimpleHashInit(tSimpleHashGetSize(pVg), taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));

×

704

  QUERY_CHECK_NULL(pExc->pBatchs, code, line, _return, terrno)

×

706

  tSimpleHashSetFreeFp(pExc->pBatchs, freeExchangeGetBasicOperatorParam);

×

708

  int32_t iter = 0;

×

709

  void*   p = NULL;

×

710

  while (NULL != (p = tSimpleHashIterate(pVg, p, &iter))) {

×

711

    int32_t* pVgId = tSimpleHashGetKey(p, NULL);

×

712

    SArray*  pUidList = *(SArray**)p;

×

714

    code = buildExchangeOperatorBasicParam(&basic, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN,

×

717

                                           (STimeWindow){0}, NULL, false, false, false);

×

718

    QUERY_CHECK_CODE(code, line, _return);

×

720

    QRY_ERR_RET(tSimpleHashPut(pExc->pBatchs, pVgId, sizeof(*pVgId), &basic, sizeof(basic)));

×

722

    basic = (SExchangeOperatorBasicParam){0};

×

723

    qTrace("build downstreamIdx %d batch scan, vgId:%d, uidNum:%" PRId64, downstreamIdx, *pVgId, (int64_t)taosArrayGetSize(pUidList));

×

726

    taosArrayDestroy(pUidList);

×

728

    *(SArray**)p = NULL;

×

730

  *ppRes = pParam;

×

732

  return code;

×

734

_return:

×

735

  qError("failed to build batch exchange operator param, code:%d", code);

×

736

  freeOperatorParam(pParam, OP_GET_PARAM);

×

737

  freeExchangeGetBasicOperatorParam(&basic);

×

738

  return code;

×

755

static int32_t buildBatchExchangeOperatorParamForVirtual(SOperatorParam** ppRes, int32_t downstreamIdx,

×

759

  int32_t                       code = TSDB_CODE_SUCCESS;

×

760

  int32_t                       lino = 0;

×

761

  SOperatorParam*               pParam = NULL;

×

762

  SExchangeOperatorBatchParam*  pExc = NULL;

×

763

  SExchangeOperatorBasicParam   basic = {0};

×

765

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

766

  QUERY_CHECK_NULL(pParam, code, lino, _return, terrno)

×

768

  pParam->value = taosMemoryMalloc(sizeof(SExchangeOperatorParam));

×

769

  QUERY_CHECK_NULL(pParam->value, code, lino, _return, terrno)

×

771

  pExc = pParam->value;

×

772

  pExc->multiParams = true;

×

774

  pExc->pBatchs = tSimpleHashInit(taosHashGetSize(pBatchMaps), taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));

×

775

  QUERY_CHECK_NULL(pExc->pBatchs, code, lino, _return, terrno)

×

776

  tSimpleHashSetFreeFp(pExc->pBatchs, freeExchangeGetBasicOperatorParam);

×

778

  size_t keyLen = 0;

×

779

  void*  pIter = taosHashIterate(pBatchMaps, NULL);

×

780

  while (pIter != NULL) {

×

781

    SArray*          pOrgTbInfoArray = *(SArray**)pIter;

×

782

    int32_t*         vgId = (int32_t*)taosHashGetKey(pIter, &keyLen);

×

784

    code = buildExchangeOperatorBasicParam(&basic, srcOpType,

×

788

    QUERY_CHECK_CODE(code, lino, _return);

×

790

    code = tSimpleHashPut(pExc->pBatchs, vgId, sizeof(*vgId), &basic, sizeof(basic));

×

791

    QUERY_CHECK_CODE(code, lino, _return);

×

793

    basic = (SExchangeOperatorBasicParam){0};

×

794

    pIter = taosHashIterate(pBatchMaps, pIter);

×

797

  pParam->pChildren = NULL;

×

798

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;

×

799

  pParam->downstreamIdx = downstreamIdx;

×

800

  pParam->reUse = false;

×

802

  *ppRes = pParam;

×

803

  return code;

×

805

_return:

×

806

  qError("failed to build exchange operator param for vscan, code:%d", code);

×

807

  freeOperatorParam(pParam, OP_GET_PARAM);

×

808

  freeExchangeGetBasicOperatorParam(&basic);

×

809

  return code;

×

812

static int32_t buildMergeJoinOperatorParam(SOperatorParam** ppRes, bool initParam, SOperatorParam** ppChild0, SOperatorParam** ppChild1) {

×

813

  int32_t code = TSDB_CODE_SUCCESS;

×

814

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

815

  if (NULL == *ppRes) {

×

816

    code = terrno;

×

817

    return code;

×

819

  (*ppRes)->pChildren = taosArrayInit(2, POINTER_BYTES);

×

820

  if (NULL == (*ppRes)->pChildren) {

×

821

    code = terrno;

×

822

    freeOperatorParam(*ppRes, OP_GET_PARAM);

×

823

    *ppRes = NULL;

×

824

    return code;

×

826

  if (NULL == taosArrayPush((*ppRes)->pChildren, ppChild0)) {

×

827

    code = terrno;

×

828

    freeOperatorParam(*ppRes, OP_GET_PARAM);

×

829

    *ppRes = NULL;

×

830

    return code;

×

832

  *ppChild0 = NULL;

×

833

  if (NULL == taosArrayPush((*ppRes)->pChildren, ppChild1)) {

×

834

    code = terrno;

×

835

    freeOperatorParam(*ppRes, OP_GET_PARAM);

×

836

    *ppRes = NULL;

×

837

    return code;

×

839

  *ppChild1 = NULL;

×

841

  SSortMergeJoinOperatorParam* pJoin = taosMemoryMalloc(sizeof(SSortMergeJoinOperatorParam));

×

842

  if (NULL == pJoin) {

×

843

    code = terrno;

×

844

    freeOperatorParam(*ppRes, OP_GET_PARAM);

×

845

    *ppRes = NULL;

×

846

    return code;

×

849

  pJoin->initDownstream = initParam;

×

851

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN;

×

852

  (*ppRes)->value = pJoin;

×

853

  (*ppRes)->reUse = false;

×

855

  return TSDB_CODE_SUCCESS;

×

858

static int32_t buildMergeJoinNotifyOperatorParam(SOperatorParam** ppRes, SOperatorParam* pChild0, SOperatorParam* pChild1) {

×

859

  int32_t code = TSDB_CODE_SUCCESS;

×

860

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

861

  if (NULL == *ppRes) {

×

862

    code = terrno;

×

863

    freeOperatorParam(pChild0, OP_NOTIFY_PARAM);

×

864

    freeOperatorParam(pChild1, OP_NOTIFY_PARAM);

×

865

    return code;

×

867

  (*ppRes)->pChildren = taosArrayInit(2, POINTER_BYTES);

×

868

  if (NULL == *ppRes) {

×

869

    code = terrno;

×

870

    taosMemoryFreeClear(*ppRes);

×

871

    freeOperatorParam(pChild0, OP_NOTIFY_PARAM);

×

872

    freeOperatorParam(pChild1, OP_NOTIFY_PARAM);

×

873

    return code;

×

875

  if (pChild0 && NULL == taosArrayPush((*ppRes)->pChildren, &pChild0)) {

×

876

    code = terrno;

×

877

    freeOperatorParam(*ppRes, OP_NOTIFY_PARAM);

×

878

    freeOperatorParam(pChild0, OP_NOTIFY_PARAM);

×

879

    freeOperatorParam(pChild1, OP_NOTIFY_PARAM);

×

880

    *ppRes = NULL;

×

881

    return code;

×

883

  if (pChild1 && NULL == taosArrayPush((*ppRes)->pChildren, &pChild1)) {

×

884

    code = terrno;

×

885

    freeOperatorParam(*ppRes, OP_NOTIFY_PARAM);

×

886

    freeOperatorParam(pChild1, OP_NOTIFY_PARAM);

×

887

    *ppRes = NULL;

×

888

    return code;

×

891

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN;

×

892

  (*ppRes)->value = NULL;

×

893

  (*ppRes)->reUse = false;

×

895

  return TSDB_CODE_SUCCESS;

×

898

static int32_t buildBatchTableScanOperatorParam(SOperatorParam** ppRes, int32_t downstreamIdx, SSHashObj* pVg) {

×

899

  int32_t code = TSDB_CODE_SUCCESS;

×

900

  int32_t vgNum = tSimpleHashGetSize(pVg);

×

901

  if (vgNum <= 0 || vgNum > 1) {

×

902

    qError("Invalid vgroup num %d to build table scan operator param", vgNum);

×

903

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

906

  int32_t iter = 0;

×

907

  void* p = NULL;

×

908

  while (NULL != (p = tSimpleHashIterate(pVg, p, &iter))) {

×

909

    SArray* pUidList = *(SArray**)p;

×

911

    code = buildTableScanOperatorParam(ppRes, pUidList, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false);

×

912

    if (code) {

×

913

      return code;

×

915

    taosArrayDestroy(pUidList);

×

916

    *(SArray**)p = NULL;

×

919

  return TSDB_CODE_SUCCESS;

×

922

static int32_t buildSingleTableScanOperatorParam(SOperatorParam** ppRes, int32_t downstreamIdx, int32_t* pVgId, int64_t* pUid) {

×

923

  SArray* pUidList = taosArrayInit(1, sizeof(int64_t));

×

924

  if (NULL == pUidList) {

×

925

    return terrno;

×

927

  if (NULL == taosArrayPush(pUidList, pUid)) {

×

928

    return terrno;

×

931

  int32_t code = buildTableScanOperatorParam(ppRes, pUidList, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, true);

×

932

  taosArrayDestroy(pUidList);

×

933

  if (code) {

×

934

    return code;

×

937

  return TSDB_CODE_SUCCESS;

×

940

static int32_t buildSeqStbJoinOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SStbJoinPrevJoinCtx* pPrev, SStbJoinPostJoinCtx* pPost, SOperatorParam** ppParam) {

×

941

  int64_t                     rowIdx = pPrev->pListHead->readIdx;

×

942

  SOperatorParam*             pSrcParam0 = NULL;

×

943

  SOperatorParam*             pSrcParam1 = NULL;

×

944

  SOperatorParam*             pGcParam0 = NULL;

×

945

  SOperatorParam*             pGcParam1 = NULL;

×

946

  int32_t*                    leftVg = pPrev->pListHead->pLeftVg + rowIdx;

×

947

  int64_t*                    leftUid = pPrev->pListHead->pLeftUid + rowIdx;

×

948

  int32_t*                    rightVg = pPrev->pListHead->pRightVg + rowIdx;

×

949

  int64_t*                    rightUid = pPrev->pListHead->pRightUid + rowIdx;

×

950

  int32_t                     code = TSDB_CODE_SUCCESS;

×

952

  qDebug("start %" PRId64 ":%" PRId64 "th stbJoin, left:%d,%" PRIu64 " - right:%d,%" PRIu64,

×

955

  QRY_ERR_RET(updatePostJoinCurrTableInfo(&pInfo->stbJoin));

×

957

  if (pInfo->stbJoin.basic.batchFetch) {

×

958

    if (pPrev->leftHash) {

×

959

      code = pInfo->stbJoin.basic.srcScan[0] ? buildBatchTableScanOperatorParam(&pSrcParam0, 0, pPrev->leftHash) : buildBatchExchangeOperatorParam(&pSrcParam0, 0, pPrev->leftHash);

×

960

      if (TSDB_CODE_SUCCESS == code) {

×

961

        code = pInfo->stbJoin.basic.srcScan[1] ? buildBatchTableScanOperatorParam(&pSrcParam1, 1, pPrev->rightHash) : buildBatchExchangeOperatorParam(&pSrcParam1, 1, pPrev->rightHash);

×

963

      if (TSDB_CODE_SUCCESS == code) {

×

964

        tSimpleHashCleanup(pPrev->leftHash);

×

965

        tSimpleHashCleanup(pPrev->rightHash);

×

966

        pPrev->leftHash = NULL;

×

967

        pPrev->rightHash = NULL;

×

971

    code = pInfo->stbJoin.basic.srcScan[0] ? buildSingleTableScanOperatorParam(&pSrcParam0, 0, leftVg, leftUid) : buildExchangeOperatorParam(&pSrcParam0, 0, leftVg, leftUid);

×

972

    if (TSDB_CODE_SUCCESS == code) {

×

973

      code = pInfo->stbJoin.basic.srcScan[1] ? buildSingleTableScanOperatorParam(&pSrcParam1, 1, rightVg, rightUid) : buildExchangeOperatorParam(&pSrcParam1, 1, rightVg, rightUid);

×

977

  bool initParam = pSrcParam0 ? true : false;

×

978

  if (TSDB_CODE_SUCCESS == code) {

×

979

    code = buildGroupCacheOperatorParam(&pGcParam0, 0, *leftVg, *leftUid, pPost->leftNeedCache, pSrcParam0);

×

980

    pSrcParam0 = NULL;

×

982

  if (TSDB_CODE_SUCCESS == code) {

×

983

    code = buildGroupCacheOperatorParam(&pGcParam1, 1, *rightVg, *rightUid, pPost->rightNeedCache, pSrcParam1);

×

984

    pSrcParam1 = NULL;

×

986

  if (TSDB_CODE_SUCCESS == code) {

×

987

    code = buildMergeJoinOperatorParam(ppParam, initParam, &pGcParam0, &pGcParam1);

×

989

  if (TSDB_CODE_SUCCESS != code) {

×

990

    if (pSrcParam0) {

×

991

      freeOperatorParam(pSrcParam0, OP_GET_PARAM);

×

993

    if (pSrcParam1) {

×

994

      freeOperatorParam(pSrcParam1, OP_GET_PARAM);

×

996

    if (pGcParam0) {

×

997

      freeOperatorParam(pGcParam0, OP_GET_PARAM);

×

999

    if (pGcParam1) {

×

1000

      freeOperatorParam(pGcParam1, OP_GET_PARAM);

×

1002

    if (*ppParam) {

×

1003

      freeOperatorParam(*ppParam, OP_GET_PARAM);

×

1004

      *ppParam = NULL;

×

1008

  return code;

×

1011

static int32_t buildVtbScanOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes, uint64_t uid) {

×

1012

  int32_t                   code = TSDB_CODE_SUCCESS;

×

1013

  int32_t                   lino = 0;

×

1014

  SVTableScanOperatorParam* pVScan = NULL;

×

1015

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

1016

  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno)

×

1018

  (*ppRes)->pChildren = taosArrayInit(1, POINTER_BYTES);

×

1019

  QUERY_CHECK_NULL((*ppRes)->pChildren, code, lino, _return, terrno)

×

1021

  pVScan = taosMemoryMalloc(sizeof(SVTableScanOperatorParam));

×

1022

  QUERY_CHECK_NULL(pVScan, code, lino, _return, terrno)

×

1023

  pVScan->pOpParamArray = taosArrayInit(1, POINTER_BYTES);

×

1024

  QUERY_CHECK_NULL(pVScan->pOpParamArray, code, lino, _return, terrno)

×

1025

  pVScan->uid = uid;

×

1026

  pVScan->window = pInfo->vtbScan.window;

×

1027

  if (pInfo->vtbScan.refColGroups) {

×

1028

    pVScan->pRefColGroups = taosArrayInit(taosArrayGetSize(pInfo->vtbScan.refColGroups), sizeof(SRefColIdGroup));

×

1029

    QUERY_CHECK_NULL(pVScan->pRefColGroups, code, lino, _return, terrno)

×

1030

    for (int32_t i = 0; i < taosArrayGetSize(pInfo->vtbScan.refColGroups); i++) {

×

1031

      SRefColIdGroup* pSrc = (SRefColIdGroup*)taosArrayGet(pInfo->vtbScan.refColGroups, i);

×

1032

      SRefColIdGroup  dst = {0};

×

1033

      QUERY_CHECK_NULL(pSrc, code, lino, _return, terrno)

×

1034

      dst.pSlotIdList = taosArrayDup(pSrc->pSlotIdList, NULL);

×

1035

      QUERY_CHECK_NULL(dst.pSlotIdList, code, lino, _return, terrno)

×

1036

      void* px = taosArrayPush(pVScan->pRefColGroups, &dst);

×

1037

      if (NULL == px) {

×

1038

        taosArrayDestroy(dst.pSlotIdList);

×

1039

        dst.pSlotIdList = NULL;

×

1041

      QUERY_CHECK_NULL(px, code, lino, _return, terrno)

×

1044

    pVScan->pRefColGroups = NULL;

×

1047

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_VIRTUAL_TABLE_SCAN;

×

1048

  (*ppRes)->downstreamIdx = 0;

×

1049

  (*ppRes)->value = pVScan;

×

1050

  (*ppRes)->reUse = false;

×

1052

  return TSDB_CODE_SUCCESS;

×

1053

_return:

×

1054

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

×

1055

  if (pVScan) {

×

1056

    if (pVScan->pRefColGroups) {

×

1057

      taosArrayDestroyEx(pVScan->pRefColGroups, destroyRefColIdGroup);

×

1058

      pVScan->pRefColGroups = NULL;

×

1060

    taosArrayDestroy(pVScan->pOpParamArray);

×

1061

    taosMemoryFreeClear(pVScan);

×

1063

  if (*ppRes) {

×

1064

    taosArrayDestroy((*ppRes)->pChildren);

×

1065

    taosMemoryFreeClear(*ppRes);

×

1067

  return code;

×

1070

static int32_t addRefColIdToRefMap(SHashObj* refMap, const char* colrefName, col_id_t colId) {

×

1071

  int32_t  code = TSDB_CODE_SUCCESS;

×

1072

  int32_t  line = 0;

×

1073

  SArray** sameRefColIdList = NULL;

×

1075

  if (colrefName == NULL || colrefName[0] == '\0') {

×

1076

    return code;

×

1079

  sameRefColIdList = (SArray**)taosHashGet(refMap, colrefName, strlen(colrefName));

×

1080

  if (sameRefColIdList == NULL) {

×

1081

    SArray* list = taosArrayInit(2, sizeof(col_id_t));

×

1082

    QUERY_CHECK_NULL(list, code, line, _return, terrno)

×

1083

    QUERY_CHECK_CODE(taosHashPut(refMap, colrefName, strlen(colrefName), &list, POINTER_BYTES), line, _return);

×

1084

    sameRefColIdList = (SArray**)taosHashGet(refMap, colrefName, strlen(colrefName));

×

1085

    QUERY_CHECK_NULL(sameRefColIdList, code, line, _return, terrno)

×

1088

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

×

1089

    col_id_t existing = *(col_id_t*)taosArrayGet(*sameRefColIdList, i);

×

1090

    if (existing == colId) {

×

1091

      return code;

×

1094

  QUERY_CHECK_NULL(taosArrayPush(*sameRefColIdList, &colId), code, line, _return, terrno)

×

1095

  return code;

×

1097

_return:

×

1098

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

×

1099

  return code;

×

1102

static int32_t buildRefSlotGroupsFromRefMap(SHashObj* refMap, SArray* readColList, SArray** ppGroups) {

×

1103

  int32_t   code = TSDB_CODE_SUCCESS;

×

1104

  int32_t   line = 0;

×

1105

  SArray*   groups = NULL;

×

1106

  SHashObj* colIdToSlot = NULL;

×

1108

  if (refMap == NULL || readColList == NULL) {

×

1109

    return code;

×

1112

  if (*ppGroups) {

×

1113

    taosArrayDestroyEx(*ppGroups, destroyRefColIdGroup);

×

1114

    *ppGroups = NULL;

×

1117

  colIdToSlot = taosHashInit(taosArrayGetSize(readColList), taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false,

×

1119

  QUERY_CHECK_NULL(colIdToSlot, code, line, _return, terrno)

×

1122

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

×

1123

    col_id_t colId = *(col_id_t*)taosArrayGet(readColList, i);

×

1124

    int32_t  slotId = i;

×

1125

    code = taosHashPut(colIdToSlot, &colId, sizeof(colId), &slotId, sizeof(slotId));

×

1126

    QUERY_CHECK_CODE(code, line, _return);

×

1129

  groups = taosArrayInit(1, sizeof(SRefColIdGroup));

×

1130

  QUERY_CHECK_NULL(groups, code, line, _return, terrno)

×

1133

  void* pIter = taosHashIterate(refMap, NULL);

×

1134

  while (pIter != NULL) {

×

1135

    SArray* pList = *(SArray**)pIter;  // colId list

×

1136

    if (pList && taosArrayGetSize(pList) > 1) {

×

1137

      SArray* slotList = taosArrayInit(taosArrayGetSize(pList), sizeof(int32_t));

×

1138

      QUERY_CHECK_NULL(slotList, code, line, _return, terrno)

×

1139

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

×

1140

        col_id_t colId = *(col_id_t*)taosArrayGet(pList, i);

×

1141

        int32_t* slotId = taosHashGet(colIdToSlot, &colId, sizeof(colId));

×

1142

        if (slotId) {

×

1143

          QUERY_CHECK_NULL(taosArrayPush(slotList, slotId), code, line, _return, terrno)

×

1146

      if (taosArrayGetSize(slotList) > 1) {

×

1147

        SRefColIdGroup g = {.pSlotIdList = slotList};

×

1148

        QUERY_CHECK_NULL(taosArrayPush(groups, &g), code, line, _return, terrno)

×

1150

        taosArrayDestroy(slotList);

×

1153

    if (pList) {

×

1154

      taosArrayDestroy(pList);

×

1156

    pIter = taosHashIterate(refMap, pIter);

×

1159

  if (taosArrayGetSize(groups) == 0) {

×

1160

    taosArrayDestroy(groups);

×

1161

    groups = NULL;

×

1164

_return:

×

1165

  if (code) {

×

1166

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

×

1167

    if (groups) {

×

1168

      taosArrayDestroyEx(groups, destroyRefColIdGroup);

×

1171

  if (refMap) {

×

1172

    taosHashCleanup(refMap);

×

1174

  if (colIdToSlot) {

×

1175

    taosHashCleanup(colIdToSlot);

×

1177

  *ppGroups = groups;

×

1178

  return code;

×

1181

bool colNeedScan(SOperatorInfo* pOperator, col_id_t colId) {

×

1182

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1183

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

1185

  if (pVtbScan->scanAllCols) {

×

1186

    return true;

×

1190

  if (pVtbScan->readColSet) {

×

1191

    return taosHashGet(pVtbScan->readColSet, &colId, sizeof(colId)) != NULL;

×

1194

  for (int32_t i = 0; i < taosArrayGetSize(pVtbScan->readColList); i++) {

×

1195

    if (colId == *(col_id_t*)taosArrayGet(pVtbScan->readColList, i)) {

×

1196

      return true;

×

1199

  return false;

×

1202

static int32_t buildExternalWindowOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes, SArray* pWins, int32_t idx) {

×

1203

  int32_t                       code = TSDB_CODE_SUCCESS;

×

1204

  int32_t                       lino = 0;

×

1205

  SExternalWindowOperatorParam* pExtWinOp = NULL;

×

1207

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

1208

  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno)

×

1210

  pExtWinOp = taosMemoryMalloc(sizeof(SExternalWindowOperatorParam));

×

1211

  QUERY_CHECK_NULL(pExtWinOp, code, lino, _return, terrno)

×

1213

  pExtWinOp->ExtWins = taosArrayDup(pWins, NULL);

×

1214

  QUERY_CHECK_NULL(pExtWinOp->ExtWins, code, lino, _return, terrno)

×

1216

  SExtWinTimeWindow *firstWin = (SExtWinTimeWindow *)taosArrayGet(pWins, 0);

×

1217

  SExtWinTimeWindow *lastWin = (SExtWinTimeWindow *)taosArrayGet(pWins, taosArrayGetSize(pWins) - 1);

×

1219

  QUERY_CHECK_NULL(firstWin, code, lino, _return, terrno)

×

1220

  QUERY_CHECK_NULL(lastWin, code, lino, _return, terrno)

×

1222

  (*ppRes)->pChildren = taosArrayInit(1, POINTER_BYTES);

×

1223

  QUERY_CHECK_NULL((*ppRes)->pChildren, code, lino, _return, terrno)

×

1225

  SOperatorParam* pExchangeOperator = NULL;

×

1226

  STimeWindow     twin = {.skey = firstWin->tw.skey, .ekey = lastWin->tw.ekey};

×

1227

  code = buildExchangeOperatorParamForExternalWindow(&pExchangeOperator, 0, twin);

×

1228

  QUERY_CHECK_CODE(code, lino, _return);

×

1229

  QUERY_CHECK_NULL(taosArrayPush((*ppRes)->pChildren, &pExchangeOperator), code, lino, _return, terrno)

×

1231

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_EXTERNAL_WINDOW;

×

1232

  (*ppRes)->downstreamIdx = idx;

×

1233

  (*ppRes)->value = pExtWinOp;

×

1234

  (*ppRes)->reUse = false;

×

1236

  return code;

×

1237

_return:

×

1238

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

×

1239

  if (pExtWinOp) {

×

1240

    if (pExtWinOp->ExtWins) {

×

1241

      taosArrayDestroy(pExtWinOp->ExtWins);

×

1243

    taosMemoryFree(pExtWinOp);

×

1245

  if (*ppRes) {

×

1246

    if ((*ppRes)->pChildren) {

×

1247

      for (int32_t i = 0; i < taosArrayGetSize((*ppRes)->pChildren); ++i) {

×

1248

        SOperatorParam* pChildParam = taosArrayGetP((*ppRes)->pChildren, i);

×

1249

        if (pChildParam) {

×

1250

          freeOperatorParam(pChildParam, OP_GET_PARAM);

×

1253

      taosArrayDestroy((*ppRes)->pChildren);

×

1255

    taosMemoryFree(*ppRes);

×

1256

    *ppRes = NULL;

×

1258

  return code;

×

1261

static int32_t buildMergeOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes, SArray* pWins,

×

1263

  int32_t                   code = TSDB_CODE_SUCCESS;

×

1264

  int32_t                   lino = 0;

×

1265

  SMergeOperatorParam*      pMergeOp = NULL;

×

1267

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

1268

  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno)

×

1270

  (*ppRes)->pChildren = taosArrayInit(numOfDownstream, POINTER_BYTES);

×

1271

  QUERY_CHECK_NULL((*ppRes)->pChildren, code, lino, _return, terrno)

×

1273

  pMergeOp = taosMemoryMalloc(sizeof(SMergeOperatorParam));

×

1274

  QUERY_CHECK_NULL(pMergeOp, code, lino, _return, terrno)

×

1276

  pMergeOp->winNum = numOfWins;

×

1278

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

×

1279

    SOperatorParam* pExternalWinParam = NULL;

×

1280

    code = buildExternalWindowOperatorParam(pInfo, &pExternalWinParam, pWins, i);

×

1281

    QUERY_CHECK_CODE(code, lino, _return);

×

1282

    QUERY_CHECK_NULL(taosArrayPush((*ppRes)->pChildren, &pExternalWinParam), code, lino, _return, terrno)

×

1285

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_MERGE;

×

1286

  (*ppRes)->downstreamIdx = 0;

×

1287

  (*ppRes)->value = pMergeOp;

×

1288

  (*ppRes)->reUse = false;

×

1290

  return TSDB_CODE_SUCCESS;

×

1291

_return:

×

1292

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

×

1293

  if (pMergeOp) {

×

1294

    taosMemoryFree(pMergeOp);

×

1296

  if (*ppRes) {

×

1297

    if ((*ppRes)->pChildren) {

×

1298

      for (int32_t i = 0; i < taosArrayGetSize((*ppRes)->pChildren); i++) {

×

1299

        SOperatorParam* pChildParam = taosArrayGetP((*ppRes)->pChildren, i);

×

1300

        if (pChildParam) {

×

1301

          freeOperatorParam(pChildParam, OP_GET_PARAM);

×

1304

      taosArrayDestroy((*ppRes)->pChildren);

×

1306

    taosMemoryFree(*ppRes);

×

1307

    *ppRes = NULL;

×

1309

  return code;

×

1320

static int32_t buildMergeOperatorParamForTsScan(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes) {

×

1321

  int32_t                   code = TSDB_CODE_SUCCESS;

×

1322

  int32_t                   lino = 0;

×

1323

  SOperatorParam*           pParam = NULL;

×

1324

  SOperatorParam*           pExchangeParam = NULL;

×

1325

  SVtbScanDynCtrlInfo*      pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

1327

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

1328

  QUERY_CHECK_NULL(pParam, code, lino, _return, terrno)

×

1330

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_MERGE;

×

1331

  pParam->downstreamIdx = 0;

×

1332

  pParam->reUse = false;

×

1333

  pParam->pChildren = taosArrayInit(1, POINTER_BYTES);

×

1334

  QUERY_CHECK_NULL(pParam->pChildren, code, lino, _return, terrno)

×

1336

  pParam->value = taosMemoryMalloc(sizeof(SMergeOperatorParam));

×

1337

  QUERY_CHECK_NULL(pParam->value, code, lino, _return, terrno)

×

1339

  for (int32_t i = 0; i < taosHashGetSize(pVtbScan->otbVgIdToOtbInfoArrayMap); i++) {

×

1340

    code = buildBatchExchangeOperatorParamForVirtual(&pExchangeParam, i, NULL, 0, pVtbScan->otbVgIdToOtbInfoArrayMap,

×

1341

                                                     (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MIN},

×

1344

    QUERY_CHECK_CODE(code, lino, _return);

×

1345

    QUERY_CHECK_NULL(taosArrayPush(pParam->pChildren, &pExchangeParam), code, lino, _return, terrno)

×

1346

    pExchangeParam = NULL;

×

1349

  *ppRes = pParam;

×

1351

  return code;

×

1352

_return:

×

1353

  if (pExchangeParam) {

×

1354

    freeOperatorParam(pExchangeParam, OP_GET_PARAM);

×

1356

  if (pParam) {

×

1357

    freeOperatorParam(pParam, OP_GET_PARAM);

×

1359

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

×

1360

  return code;

×

1363

static int32_t buildAggOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes) {

×

1364

  int32_t                   code = TSDB_CODE_SUCCESS;

×

1365

  int32_t                   lino = 0;

×

1366

  SOperatorParam*           pParam = NULL;

×

1367

  SOperatorParam*           pExchangeParam = NULL;

×

1368

  SVtbScanDynCtrlInfo*      pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

1369

  bool                      freeExchange = false;

×

1371

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

1372

  QUERY_CHECK_NULL(pParam, code, lino, _return, terrno)

×

1374

  pParam->pChildren = taosArrayInit(1, POINTER_BYTES);

×

1375

  QUERY_CHECK_NULL(pParam->pChildren, code, lino, _return, terrno)

×

1377

  pParam->value = taosMemoryMalloc(sizeof(SAggOperatorParam));

×

1378

  QUERY_CHECK_NULL(pParam->value, code, lino, _return, terrno)

×

1380

  code = buildBatchExchangeOperatorParamForVirtual(

×

1382

      (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MIN}, EX_SRC_TYPE_VSTB_AGG_SCAN,

×

1384

  QUERY_CHECK_CODE(code, lino, _return);

×

1386

  freeExchange = true;

×

1388

  QUERY_CHECK_NULL(taosArrayPush(pParam->pChildren, &pExchangeParam), code, lino, _return, terrno)

×

1390

  freeExchange = false;

×

1392

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_HASH_AGG;

×

1393

  pParam->downstreamIdx = 0;

×

1394

  pParam->reUse = false;

×

1396

  *ppRes = pParam;

×

1398

  return code;

×

1399

_return:

×

1400

  if (freeExchange) {

×

1401

    freeOperatorParam(pExchangeParam, OP_GET_PARAM);

×

1403

  if (pParam) {

×

1404

    freeOperatorParam(pParam, OP_GET_PARAM);

×

1406

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

×

1407

  return code;

×

1410

static int32_t buildAggOperatorParamWithGroupId(SDynQueryCtrlOperatorInfo* pInfo, uint64_t groupid,

×

1412

  int32_t                   code = TSDB_CODE_SUCCESS;

×

1413

  int32_t                   lino = 0;

×

1414

  SVtbScanDynCtrlInfo*      pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

1415

  SOperatorParam*           pParam = NULL;

×

1416

  SOperatorParam*           pExchangeParam = NULL;

×

1417

  SHashObj*                 otbVgIdToOtbInfoArrayMap = NULL;

×

1418

  bool                      freeExchange = false;

×

1419

  void*                     pIter = taosHashGet(pVtbScan->vtbGroupIdToVgIdMapMap, &groupid, sizeof(groupid));

×

1421

  if (!pIter) {

×

1422

    *ppRes = NULL;

×

1423

    return code;

×

1426

  otbVgIdToOtbInfoArrayMap = *(SHashObj**)pIter;

×

1428

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

1429

  QUERY_CHECK_NULL(pParam, code, lino, _return, terrno)

×

1431

  pParam->pChildren = taosArrayInit(1, POINTER_BYTES);

×

1432

  QUERY_CHECK_NULL(pParam->pChildren, code, lino, _return, terrno)

×

1434

  code = buildBatchExchangeOperatorParamForVirtual(

×

1436

      (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MIN}, EX_SRC_TYPE_VSTB_AGG_SCAN,

×

1438

  QUERY_CHECK_CODE(code, lino, _return);

×

1440

  freeExchange = true;

×

1442

  QUERY_CHECK_NULL(taosArrayPush(pParam->pChildren, &pExchangeParam), code, lino, _return, terrno)

×

1444

  freeExchange = false;

×

1446

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_HASH_AGG;

×

1447

  pParam->downstreamIdx = 0;

×

1448

  pParam->value = NULL;

×

1449

  pParam->reUse = false;

×

1451

  *ppRes = pParam;

×

1453

  return code;

×

1454

_return:

×

1455

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

×

1456

  if (freeExchange) {

×

1457

    freeOperatorParam(pExchangeParam, OP_GET_PARAM);

×

1459

  if (pParam) {

×

1460

    freeOperatorParam(pParam, OP_GET_PARAM);

×

1462

  return code;

×

1465

static int32_t buildAggOperatorParamForSingleChild(SDynQueryCtrlOperatorInfo* pInfo, tb_uid_t uid,

×

1467

  int32_t                   code = TSDB_CODE_SUCCESS;

×

1468

  int32_t                   lino = 0;

×

1469

  SVtbScanDynCtrlInfo*      pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

1470

  SOperatorParam*           pParam = NULL;

×

1471

  SHashObj*                 pOtbVgIdToOtbInfoArrayMap = NULL;

×

1472

  void*                     pIter = taosHashGet(pVtbScan->vtbUidToVgIdMapMap, &uid, sizeof(uid));

×

1474

  if (pIter) {

×

1475

    pOtbVgIdToOtbInfoArrayMap = *(SHashObj**)taosHashGet(pVtbScan->vtbUidToVgIdMapMap, &uid, sizeof(uid));

×

1477

    code = buildBatchExchangeOperatorParamForVirtual(

×

1479

        (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MIN}, EX_SRC_TYPE_VSTB_AGG_SCAN,

×

1481

    QUERY_CHECK_CODE(code, lino, _return);

×

1483

    *ppRes = pParam;

×

1485

    *ppRes = NULL;

×

1488

  return code;

×

1489

_return:

×

1490

  if (pParam) {

×

1491

    freeOperatorParam(pParam, OP_GET_PARAM);

×

1493

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

×

1494

  return code;

×

1497

static void seqJoinLaunchNewRetrieveImpl(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

1498

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1499

  SStbJoinDynCtrlInfo*       pStbJoin = (SStbJoinDynCtrlInfo*)&pInfo->stbJoin;

×

1500

  SStbJoinPrevJoinCtx*       pPrev = &pStbJoin->ctx.prev;

×

1501

  SStbJoinPostJoinCtx*       pPost = &pStbJoin->ctx.post;

×

1502

  SOperatorParam*            pParam = NULL;

×

1503

  int32_t                    code  = buildSeqStbJoinOperatorParam(pInfo, pPrev, pPost, &pParam);

×

1504

  if (TSDB_CODE_SUCCESS != code) {

×

1505

    pOperator->pTaskInfo->code = code;

×

1506

    T_LONG_JMP(pOperator->pTaskInfo->env, pOperator->pTaskInfo->code);

×

1509

  qDebug("%s dynamic post task begin", GET_TASKID(pOperator->pTaskInfo));

×

1510

  code = pOperator->pDownstream[1]->fpSet.getNextExtFn(pOperator->pDownstream[1], pParam, ppRes);

×

1511

  if (*ppRes && (code == 0)) {

×

1512

    code = blockDataCheck(*ppRes);

×

1513

    if (code) {

×

1514

      qError("Invalid block data, blockDataCheck failed, error:%s", tstrerror(code));

×

1515

      pOperator->pTaskInfo->code = code;

×

1516

      T_LONG_JMP(pOperator->pTaskInfo->env, pOperator->pTaskInfo->code);

×

1518

    pPost->isStarted = true;

×

1519

    pStbJoin->execInfo.postBlkNum++;

×

1520

    pStbJoin->execInfo.postBlkRows += (*ppRes)->info.rows;

×

1521

    qDebug("%s join res block retrieved", GET_TASKID(pOperator->pTaskInfo));

×

1523

    qDebug("%s Empty join res block retrieved", GET_TASKID(pOperator->pTaskInfo));

×

1525

×

1528

static int32_t notifySeqJoinTableCacheEnd(SOperatorInfo* pOperator, SStbJoinPostJoinCtx* pPost, bool leftTable) {

×

1529

  SOperatorParam* pGcParam = NULL;

×

1530

  SOperatorParam* pMergeJoinParam = NULL;

×

1531

  int32_t         downstreamId = leftTable ? 0 : 1;

×

1532

  int32_t         vgId = leftTable ? pPost->leftVgId : pPost->rightVgId;

×

1533

  int64_t         uid = leftTable ? pPost->leftCurrUid : pPost->rightCurrUid;

×

1535

  qDebug("notify table %" PRIu64 " in vgId %d downstreamId %d cache end", uid, vgId, downstreamId);

×

1537

  int32_t code = buildGroupCacheNotifyOperatorParam(&pGcParam, downstreamId, vgId, uid);

×

1538

  if (TSDB_CODE_SUCCESS != code) {

×

1539

    return code;

×

1541

  code = buildMergeJoinNotifyOperatorParam(&pMergeJoinParam, pGcParam, NULL);

×

1542

  if (TSDB_CODE_SUCCESS != code) {

×

1543

    return code;

×

1546

  return optrDefaultNotifyFn(pOperator->pDownstream[1], pMergeJoinParam);

×

1549

static int32_t handleSeqJoinCurrRetrieveEnd(SOperatorInfo* pOperator, SStbJoinDynCtrlInfo*          pStbJoin) {

×

1550

  SStbJoinPostJoinCtx* pPost = &pStbJoin->ctx.post;

×

1551

  int32_t code = 0;

×

1553

  pPost->isStarted = false;

×

1555

  if (pStbJoin->basic.batchFetch) {

×

1556

    return TSDB_CODE_SUCCESS;

×

1559

  if (pPost->leftNeedCache) {

×

1560

    uint32_t* num = tSimpleHashGet(pStbJoin->ctx.prev.leftCache, &pPost->leftCurrUid, sizeof(pPost->leftCurrUid));

×

1561

    if (num && --(*num) <= 0) {

×

1562

      code = tSimpleHashRemove(pStbJoin->ctx.prev.leftCache, &pPost->leftCurrUid, sizeof(pPost->leftCurrUid));

×

1563

      if (code) {

×

1564

        qError("tSimpleHashRemove leftCurrUid %" PRId64 " from leftCache failed, error:%s", pPost->leftCurrUid, tstrerror(code));

×

1565

        QRY_ERR_RET(code);

×

1567

      QRY_ERR_RET(notifySeqJoinTableCacheEnd(pOperator, pPost, true));

×

1571

  if (!pPost->rightNeedCache) {

×

1572

    void* v = tSimpleHashGet(pStbJoin->ctx.prev.rightCache, &pPost->rightCurrUid, sizeof(pPost->rightCurrUid));

×

1573

    if (NULL != v) {

×

1574

      code = tSimpleHashRemove(pStbJoin->ctx.prev.rightCache, &pPost->rightCurrUid, sizeof(pPost->rightCurrUid));

×

1575

      if (code) {

×

1576

        qError("tSimpleHashRemove rightCurrUid %" PRId64 " from rightCache failed, error:%s", pPost->rightCurrUid, tstrerror(code));

×

1577

        QRY_ERR_RET(code);

×

1579

      QRY_ERR_RET(notifySeqJoinTableCacheEnd(pOperator, pPost, false));

×

1583

  return TSDB_CODE_SUCCESS;

×

1588

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1589

  SStbJoinPostJoinCtx*       pPost = &pInfo->stbJoin.ctx.post;

×

1590

  SStbJoinPrevJoinCtx*       pPrev = &pInfo->stbJoin.ctx.prev;

×

1592

  if (!pPost->isStarted) {

×

1593

    return TSDB_CODE_SUCCESS;

×

1596

  qDebug("%s dynQueryCtrl continue to retrieve block from post op", GET_TASKID(pOperator->pTaskInfo));

×

1598

  *ppRes = getNextBlockFromDownstream(pOperator, 1);

×

1599

  if (NULL == *ppRes) {

×

1600

    QRY_ERR_RET(handleSeqJoinCurrRetrieveEnd(pOperator, &pInfo->stbJoin));

×

1601

    pPrev->pListHead->readIdx++;

×

1603

    pInfo->stbJoin.execInfo.postBlkNum++;

×

1604

    pInfo->stbJoin.execInfo.postBlkRows += (*ppRes)->info.rows;

×

1607

  return TSDB_CODE_SUCCESS;

×

1611

  SArray** ppArray = tSimpleHashGet(pHash, pKey, keySize);

×

1612

  if (NULL == ppArray) {

×

1613

    SArray* pArray = taosArrayInit(10, valSize);

×

1614

    if (NULL == pArray) {

×

1615

      return terrno;

×

1617

    if (NULL == taosArrayPush(pArray, pVal)) {

×

1618

      taosArrayDestroy(pArray);

×

1619

      return terrno;

×

1621

    if (tSimpleHashPut(pHash, pKey, keySize, &pArray, POINTER_BYTES)) {

×

1622

      taosArrayDestroy(pArray);

×

1623

      return terrno;

×

1625

    return TSDB_CODE_SUCCESS;

×

1628

  if (NULL == taosArrayPush(*ppArray, pVal)) {

×

1629

    return terrno;

×

1632

  return TSDB_CODE_SUCCESS;

×

1636

  int32_t code = TSDB_CODE_SUCCESS;

×

1637

  uint32_t* pNum = tSimpleHashGet(pHash, pKey, keySize);

×

1638

  if (NULL == pNum) {

×

1639

    uint32_t n = 1;

×

1640

    code = tSimpleHashPut(pHash, pKey, keySize, &n, sizeof(n));

×

1641

    if (code) {

×

1642

      return code;

×

1644

    code = tSimpleHashPut(pOnceHash, pKey, keySize, NULL, 0);

×

1645

    if (code) {

×

1646

      return code;

×

1648

    return TSDB_CODE_SUCCESS;

×

1651

  switch (*pNum) {

×

1652

    case 0:

×

1653

      break;

×

1654

    case UINT32_MAX:

×

1655

      *pNum = 0;

×

1656

      break;

×

1657

    default:

×

1658

      if (1 == (*pNum)) {

×

1659

        code = tSimpleHashRemove(pOnceHash, pKey, keySize);

×

1660

        if (code) {

×

1661

          qError("tSimpleHashRemove failed in addToJoinTableHash, error:%s", tstrerror(code));

×

1662

          QRY_ERR_RET(code);

×

1665

      (*pNum)++;

×

1666

      break;

×

1669

  return TSDB_CODE_SUCCESS;

×

1673

static void freeStbJoinTableList(SStbJoinTableList* pList) {

×

1674

  if (NULL == pList) {

×

1675

    return;

×

1677

  taosMemoryFree(pList->pLeftVg);

×

1678

  taosMemoryFree(pList->pLeftUid);

×

1679

  taosMemoryFree(pList->pRightVg);

×

1680

  taosMemoryFree(pList->pRightUid);

×

1681

  taosMemoryFree(pList);

×

1684

static int32_t appendStbJoinTableList(SStbJoinPrevJoinCtx* pCtx, int64_t rows, int32_t* pLeftVg, int64_t* pLeftUid, int32_t* pRightVg, int64_t* pRightUid) {

×

1685

  int32_t code = TSDB_CODE_SUCCESS;

×

1686

  SStbJoinTableList* pNew = taosMemoryCalloc(1, sizeof(SStbJoinTableList));

×

1687

  if (NULL == pNew) {

×

1688

    return terrno;

×

1690

  pNew->pLeftVg = taosMemoryMalloc(rows * sizeof(*pLeftVg));

×

1691

  if (NULL == pNew->pLeftVg) {

×

1692

    code = terrno;

×

1693

    freeStbJoinTableList(pNew);

×

1694

    return code;

×

1696

  pNew->pLeftUid = taosMemoryMalloc(rows * sizeof(*pLeftUid));

×

1697

  if (NULL == pNew->pLeftUid) {

×

1698

    code = terrno;

×

1699

    freeStbJoinTableList(pNew);

×

1700

    return code;

×

1702

  pNew->pRightVg = taosMemoryMalloc(rows * sizeof(*pRightVg));

×

1703

  if (NULL == pNew->pRightVg) {

×

1704

    code = terrno;

×

1705

    freeStbJoinTableList(pNew);

×

1706

    return code;

×

1708

  pNew->pRightUid = taosMemoryMalloc(rows * sizeof(*pRightUid));

×

1709

  if (NULL == pNew->pRightUid) {

×

1710

    code = terrno;

×

1711

    freeStbJoinTableList(pNew);

×

1712

    return code;

×

1715

  TAOS_MEMCPY(pNew->pLeftVg, pLeftVg, rows * sizeof(*pLeftVg));

×

1716

  TAOS_MEMCPY(pNew->pLeftUid, pLeftUid, rows * sizeof(*pLeftUid));

×

1717

  TAOS_MEMCPY(pNew->pRightVg, pRightVg, rows * sizeof(*pRightVg));

×

1718

  TAOS_MEMCPY(pNew->pRightUid, pRightUid, rows * sizeof(*pRightUid));

×

1720

  pNew->readIdx = 0;

×

1721

  pNew->uidNum = rows;

×

1722

  pNew->pNext = NULL;

×

1724

  if (pCtx->pListTail) {

×

1725

    pCtx->pListTail->pNext = pNew;

×

1726

    pCtx->pListTail = pNew;

×

1728

    pCtx->pListHead = pNew;

×

1729

    pCtx->pListTail= pNew;

×

1732

  return TSDB_CODE_SUCCESS;

×

1735

static void doBuildStbJoinTableHash(SOperatorInfo* pOperator, SSDataBlock* pBlock) {

×

1736

  int32_t                    code = TSDB_CODE_SUCCESS;

×

1737

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1738

  SStbJoinDynCtrlInfo*       pStbJoin = (SStbJoinDynCtrlInfo*)&pInfo->stbJoin;

×

1739

  SColumnInfoData*           pVg0 = taosArrayGet(pBlock->pDataBlock, pStbJoin->basic.vgSlot[0]);

×

1740

  if (NULL == pVg0) {

×

1741

    QRY_ERR_JRET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

1743

  SColumnInfoData*           pVg1 = taosArrayGet(pBlock->pDataBlock, pStbJoin->basic.vgSlot[1]);

×

1744

  if (NULL == pVg1) {

×

1745

    QRY_ERR_JRET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

1747

  SColumnInfoData*           pUid0 = taosArrayGet(pBlock->pDataBlock, pStbJoin->basic.uidSlot[0]);

×

1748

  if (NULL == pUid0) {

×

1749

    QRY_ERR_JRET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

1751

  SColumnInfoData*           pUid1 = taosArrayGet(pBlock->pDataBlock, pStbJoin->basic.uidSlot[1]);

×

1752

  if (NULL == pUid1) {

×

1753

    QRY_ERR_JRET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

1756

  if (pStbJoin->basic.batchFetch) {

×

1757

    for (int32_t i = 0; i < pBlock->info.rows; ++i) {

×

1758

      int32_t* leftVg = (int32_t*)(pVg0->pData + pVg0->info.bytes * i);

×

1759

      int64_t* leftUid = (int64_t*)(pUid0->pData + pUid0->info.bytes * i);

×

1760

      int32_t* rightVg = (int32_t*)(pVg1->pData + pVg1->info.bytes * i);

×

1761

      int64_t* rightUid = (int64_t*)(pUid1->pData + pUid1->info.bytes * i);

×

1763

      code = addToJoinVgroupHash(pStbJoin->ctx.prev.leftHash, leftVg, sizeof(*leftVg), leftUid, sizeof(*leftUid));

×

1764

      if (TSDB_CODE_SUCCESS != code) {

×

1765

        break;

×

1767

      code = addToJoinVgroupHash(pStbJoin->ctx.prev.rightHash, rightVg, sizeof(*rightVg), rightUid, sizeof(*rightUid));

×

1768

      if (TSDB_CODE_SUCCESS != code) {

×

1769

        break;

×

1773

    for (int32_t i = 0; i < pBlock->info.rows; ++i) {

×

1774

      int64_t* leftUid = (int64_t*)(pUid0->pData + pUid0->info.bytes * i);

×

1776

      code = addToJoinTableHash(pStbJoin->ctx.prev.leftCache, pStbJoin->ctx.prev.onceTable, leftUid, sizeof(*leftUid));

×

1777

      if (TSDB_CODE_SUCCESS != code) {

×

1778

        break;

×

1783

  if (TSDB_CODE_SUCCESS == code) {

×

1784

    code = appendStbJoinTableList(&pStbJoin->ctx.prev, pBlock->info.rows, (int32_t*)pVg0->pData, (int64_t*)pUid0->pData, (int32_t*)pVg1->pData, (int64_t*)pUid1->pData);

×

1785

    if (TSDB_CODE_SUCCESS == code) {

×

1786

      pStbJoin->ctx.prev.tableNum += pBlock->info.rows;

×

1790

_return:

×

1792

  if (TSDB_CODE_SUCCESS != code) {

×

1793

    pOperator->pTaskInfo->code = code;

×

1794

    T_LONG_JMP(pOperator->pTaskInfo->env, pOperator->pTaskInfo->code);

×

1796

×

1799

static void postProcessStbJoinTableHash(SOperatorInfo* pOperator) {

×

1800

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1801

  SStbJoinDynCtrlInfo*       pStbJoin = (SStbJoinDynCtrlInfo*)&pInfo->stbJoin;

×

1803

  if (pStbJoin->basic.batchFetch) {

×

1804

    return;

×

1807

  if (tSimpleHashGetSize(pStbJoin->ctx.prev.leftCache) == tSimpleHashGetSize(pStbJoin->ctx.prev.onceTable)) {

×

1808

    tSimpleHashClear(pStbJoin->ctx.prev.leftCache);

×

1809

    return;

×

1812

  uint64_t* pUid = NULL;

×

1813

  int32_t iter = 0;

×

1814

  int32_t code = 0;

×

1815

  while (NULL != (pUid = tSimpleHashIterate(pStbJoin->ctx.prev.onceTable, pUid, &iter))) {

×

1816

    code = tSimpleHashRemove(pStbJoin->ctx.prev.leftCache, pUid, sizeof(*pUid));

×

1817

    if (code) {

×

1818

      qError("tSimpleHashRemove failed in postProcessStbJoinTableHash, error:%s", tstrerror(code));

×

1822

  pStbJoin->execInfo.leftCacheNum = tSimpleHashGetSize(pStbJoin->ctx.prev.leftCache);

×

1823

  qDebug("more than 1 ref build table num %" PRId64, (int64_t)tSimpleHashGetSize(pStbJoin->ctx.prev.leftCache));

×

1835

static void buildStbJoinTableList(SOperatorInfo* pOperator) {

×

1836

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1837

  SStbJoinDynCtrlInfo*       pStbJoin = (SStbJoinDynCtrlInfo*)&pInfo->stbJoin;

×

1839

  while (true) {

×

1840

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

1841

    if (NULL == pBlock) {

×

1842

      break;

×

1845

    pStbJoin->execInfo.prevBlkNum++;

×

1846

    pStbJoin->execInfo.prevBlkRows += pBlock->info.rows;

×

1848

    doBuildStbJoinTableHash(pOperator, pBlock);

×

1851

  postProcessStbJoinTableHash(pOperator);

×

1853

  pStbJoin->ctx.prev.joinBuild = true;

×

1854

×

1856

static int32_t seqJoinLaunchNewRetrieve(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

1857

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1858

  SStbJoinDynCtrlInfo*       pStbJoin = (SStbJoinDynCtrlInfo*)&pInfo->stbJoin;

×

1859

  SStbJoinPrevJoinCtx*       pPrev = &pStbJoin->ctx.prev;

×

1860

  SStbJoinTableList*         pNode = pPrev->pListHead;

×

1862

  while (pNode) {

×

1863

    if (pNode->readIdx >= pNode->uidNum) {

×

1864

      pPrev->pListHead = pNode->pNext;

×

1865

      freeStbJoinTableList(pNode);

×

1866

      pNode = pPrev->pListHead;

×

1867

      continue;

×

1870

    seqJoinLaunchNewRetrieveImpl(pOperator, ppRes);

×

1871

    if (*ppRes) {

×

1872

      return TSDB_CODE_SUCCESS;

×

1875

    QRY_ERR_RET(handleSeqJoinCurrRetrieveEnd(pOperator, pStbJoin));

×

1876

    pPrev->pListHead->readIdx++;

×

1879

  *ppRes = NULL;

×

1880

  setOperatorCompleted(pOperator);

×

1882

  return TSDB_CODE_SUCCESS;

×

1885

static int32_t seqStableJoinComposeRes(SStbJoinDynCtrlInfo* pStbJoin, SSDataBlock* pBlock) {

×

1886

  if (pBlock) {

×

1887

    if (pStbJoin && pStbJoin->pOutputDataBlockDesc) {

×

1888

      pBlock->info.id.blockId = pStbJoin->pOutputDataBlockDesc->dataBlockId;

×

1889

      if (!pBlock->pDataBlock) return TSDB_CODE_SUCCESS;

×

1891

      for (int i = (int)pBlock->pDataBlock->size; i < pStbJoin->pOutputDataBlockDesc->pSlots->length; i++) {

×

1892

        SSlotDescNode* pSlot = (SSlotDescNode*)nodesListGetNode(pStbJoin->pOutputDataBlockDesc->pSlots, i);

×

1893

        if (pSlot == NULL) {

×

1894

          qError("seqStableJoinComposeRes: pSlot is NULL, i:%d", i);

×

1895

          return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1897

        SColumnInfoData colInfo = createColumnInfoData(pSlot->dataType.type, pSlot->dataType.bytes, pSlot->slotId);

×

1898

        int32_t code = colInfoDataEnsureCapacity(&colInfo, pBlock->info.rows, true);

×

1899

        if (code != TSDB_CODE_SUCCESS) {

×

1900

          return code;

×

1902

        code = blockDataAppendColInfo(pBlock, &colInfo);

×

1903

        if (code != TSDB_CODE_SUCCESS) {

×

1904

          return code;

×

1908

      qError("seqStableJoinComposeRes: pBlock or pStbJoin is NULL");

×

1909

      return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1912

  return TSDB_CODE_SUCCESS;

×

1915

int32_t seqStableJoin(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

1916

  int32_t                    code = TSDB_CODE_SUCCESS;

×

1917

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

1918

  SStbJoinDynCtrlInfo*       pStbJoin = (SStbJoinDynCtrlInfo*)&pInfo->stbJoin;

×

1920

  QRY_PARAM_CHECK(pRes);

×

1921

  if (pOperator->status == OP_EXEC_DONE) {

×

1922

    return code;

×

1925

  int64_t st = 0;

×

1926

  if (pOperator->cost.openCost == 0) {

×

1927

    st = taosGetTimestampUs();

×

1930

  if (!pStbJoin->ctx.prev.joinBuild) {

×

1931

    buildStbJoinTableList(pOperator);

×

1932

    if (pStbJoin->execInfo.prevBlkRows <= 0) {

×

1933

      setOperatorCompleted(pOperator);

×

1934

      goto _return;

×

1938

  QRY_ERR_JRET(seqJoinContinueCurrRetrieve(pOperator, pRes));

×

1939

  if (*pRes) {

×

1940

    goto _return;

×

1943

  QRY_ERR_JRET(seqJoinLaunchNewRetrieve(pOperator, pRes));

×

1945

_return:

×

1946

  if (pOperator->cost.openCost == 0) {

×

1947

    pOperator->cost.openCost = (double)(taosGetTimestampUs() - st) / 1000.0;

×

1950

  if (code) {

×

1951

    qError("%s failed since %s", __func__, tstrerror(code));

×

1952

    pOperator->pTaskInfo->code = code;

×

1953

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

1955

    code = seqStableJoinComposeRes(pStbJoin, *pRes);

×

1957

  return code;

×

1960

int32_t dynProcessUseDbRsp(void* param, SDataBuf* pMsg, int32_t code) {

×

1961

  int32_t                    lino = 0;

×

1962

  SOperatorInfo*             operator=(SOperatorInfo*) param;

×

1963

  SDynQueryCtrlOperatorInfo* pScanResInfo = (SDynQueryCtrlOperatorInfo*)operator->info;

×

1965

  if (TSDB_CODE_SUCCESS != code) {

×

1966

    operator->pTaskInfo->code = rpcCvtErrCode(code);

×

1967

    if (operator->pTaskInfo->code != code) {

×

1968

      qError("load systable rsp received, error:%s, cvted error:%s", tstrerror(code),

×

1971

      qError("load systable rsp received, error:%s", tstrerror(code));

×

1973

    goto _return;

×

1976

  pScanResInfo->vtbScan.pRsp = taosMemoryMalloc(sizeof(SUseDbRsp));

×

1977

  QUERY_CHECK_NULL(pScanResInfo->vtbScan.pRsp, code, lino, _return, terrno)

×

1979

  code = tDeserializeSUseDbRsp(pMsg->pData, (int32_t)pMsg->len, pScanResInfo->vtbScan.pRsp);

×

1980

  QUERY_CHECK_CODE(code, lino, _return);

×

1982

  taosMemoryFreeClear(pMsg->pData);

×

1984

  code = tsem_post(&pScanResInfo->vtbScan.ready);

×

1985

  QUERY_CHECK_CODE(code, lino, _return);

×

1987

  return code;

×

1988

_return:

×

1989

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

×

1990

  return code;

×

1993

static int32_t buildDbVgInfoMap(SOperatorInfo* pOperator, SMsgCb* pMsgCb, SName* name, SExecTaskInfo* pTaskInfo, SUseDbOutput* output) {

×

1994

  int32_t                    code = TSDB_CODE_SUCCESS;

×

1995

  int32_t                    lino = 0;

×

1996

  char*                      buf1 = NULL;

×

1997

  SUseDbReq*                 pReq = NULL;

×

1998

  SDynQueryCtrlOperatorInfo* pScanResInfo = (SDynQueryCtrlOperatorInfo*)pOperator->info;

×

2000

  pReq = taosMemoryMalloc(sizeof(SUseDbReq));

×

2001

  QUERY_CHECK_NULL(pReq, code, lino, _return, terrno)

×

2002

  code = tNameGetFullDbName(name, pReq->db);

×

2003

  QUERY_CHECK_CODE(code, lino, _return);

×

2004

  int32_t contLen = tSerializeSUseDbReq(NULL, 0, pReq);

×

2005

  buf1 = taosMemoryCalloc(1, contLen);

×

2006

  QUERY_CHECK_NULL(buf1, code, lino, _return, terrno)

×

2007

  int32_t tempRes = tSerializeSUseDbReq(buf1, contLen, pReq);

×

2008

  if (tempRes < 0) {

×

2009

    QUERY_CHECK_CODE(terrno, lino, _return);

×

2013

  SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));

×

2014

  QUERY_CHECK_NULL(pMsgSendInfo, code, lino, _return, terrno)

×

2016

  pMsgSendInfo->param = pOperator;

×

2017

  pMsgSendInfo->msgInfo.pData = buf1;

×

2018

  pMsgSendInfo->msgInfo.len = contLen;

×

2019

  pMsgSendInfo->msgType = TDMT_MND_GET_DB_INFO;

×

2020

  pMsgSendInfo->fp = dynProcessUseDbRsp;

×

2021

  pMsgSendInfo->requestId = pTaskInfo->id.queryId;

×

2023

  code = asyncSendMsgToServer(pMsgCb->clientRpc, &pScanResInfo->vtbScan.epSet, NULL, pMsgSendInfo);

×

2024

  QUERY_CHECK_CODE(code, lino, _return);

×

2026

  code = tsem_wait(&pScanResInfo->vtbScan.ready);

×

2027

  QUERY_CHECK_CODE(code, lino, _return);

×

2029

  code = queryBuildUseDbOutput(output, pScanResInfo->vtbScan.pRsp);

×

2030

  QUERY_CHECK_CODE(code, lino, _return);

×

2032

_return:

×

2033

  if (code) {

×

2034

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

×

2035

     taosMemoryFree(buf1);

×

2037

  taosMemoryFree(pReq);

×

2038

  tFreeSUsedbRsp(pScanResInfo->vtbScan.pRsp);

×

2039

  taosMemoryFreeClear(pScanResInfo->vtbScan.pRsp);

×

2040

  return code;

×

2043

int dynVgInfoComp(const void* lp, const void* rp) {

×

2044

  SVgroupInfo* pLeft = (SVgroupInfo*)lp;

×

2045

  SVgroupInfo* pRight = (SVgroupInfo*)rp;

×

2046

  if (pLeft->hashBegin < pRight->hashBegin) {

×

2047

    return -1;

×

2048

  } else if (pLeft->hashBegin > pRight->hashBegin) {

×

2049

    return 1;

×

2052

  return 0;

×

2055

int32_t dynMakeVgArraySortBy(SDBVgInfo* dbInfo, __compar_fn_t sort_func) {

×

2056

  if (NULL == dbInfo) {

×

2057

    return TSDB_CODE_SUCCESS;

×

2060

  if (dbInfo->vgHash && NULL == dbInfo->vgArray) {

×

2061

    int32_t vgSize = taosHashGetSize(dbInfo->vgHash);

×

2062

    dbInfo->vgArray = taosArrayInit(vgSize, sizeof(SVgroupInfo));

×

2063

    if (NULL == dbInfo->vgArray) {

×

2064

      return terrno;

×

2067

    void* pIter = taosHashIterate(dbInfo->vgHash, NULL);

×

2068

    while (pIter) {

×

2069

      if (NULL == taosArrayPush(dbInfo->vgArray, pIter)) {

×

2070

        taosHashCancelIterate(dbInfo->vgHash, pIter);

×

2071

        return terrno;

×

2074

      pIter = taosHashIterate(dbInfo->vgHash, pIter);

×

2077

    taosArraySort(dbInfo->vgArray, sort_func);

×

2080

  return TSDB_CODE_SUCCESS;

×

2083

int32_t dynHashValueComp(void const* lp, void const* rp) {

×

2084

  uint32_t*    key = (uint32_t*)lp;

×

2085

  SVgroupInfo* pVg = (SVgroupInfo*)rp;

×

2087

  if (*key < pVg->hashBegin) {

×

2088

    return -1;

×

2089

  } else if (*key > pVg->hashEnd) {

×

2090

    return 1;

×

2093

  return 0;

×

2096

int32_t getVgId(SDBVgInfo* dbInfo, char* dbFName, int32_t* vgId, char *tbName) {

×

2097

  int32_t code = 0;

×

2098

  int32_t lino = 0;

×

2099

  code = dynMakeVgArraySortBy(dbInfo, dynVgInfoComp);

×

2100

  QUERY_CHECK_CODE(code, lino, _return);

×

2102

  int32_t vgNum = (int32_t)taosArrayGetSize(dbInfo->vgArray);

×

2103

  if (vgNum <= 0) {

×

2104

    qError("db vgroup cache invalid, db:%s, vgroup number:%d", dbFName, vgNum);

×

2105

    QUERY_CHECK_CODE(code = TSDB_CODE_TSC_DB_NOT_SELECTED, lino, _return);

×

2108

  SVgroupInfo* vgInfo = NULL;

×

2110

  (void)snprintf(tbFullName, sizeof(tbFullName), "%s.", dbFName);

×

2111

  int32_t offset = (int32_t)strlen(tbFullName);

×

2113

  (void)snprintf(tbFullName + offset, sizeof(tbFullName) - offset, "%s", tbName);

×

2114

  uint32_t hashValue = taosGetTbHashVal(tbFullName, (int32_t)strlen(tbFullName), dbInfo->hashMethod,

×

2115

                                        dbInfo->hashPrefix, dbInfo->hashSuffix);

×

2117

  vgInfo = taosArraySearch(dbInfo->vgArray, &hashValue, dynHashValueComp, TD_EQ);

×

2118

  if (NULL == vgInfo) {

×

2119

    qError("no hash range found for hash value [%u], db:%s, numOfVgId:%d", hashValue, dbFName,

×

2121

    return TSDB_CODE_CTG_INTERNAL_ERROR;

×

2124

  *vgId = vgInfo->vgId;

×

2126

_return:

×

2127

  return code;

×

2130

int32_t getDbVgInfo(SOperatorInfo* pOperator, SName *name, SDBVgInfo **dbVgInfo) {

×

2131

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2132

  int32_t                    line = 0;

×

2133

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2134

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2135

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2136

  SMsgCb*                    pMsgCb = pVtbScan->pMsgCb;

×

2137

  SUseDbOutput*              output = NULL;

×

2138

  SUseDbOutput**             find = (SUseDbOutput**)taosHashGet(pInfo->vtbScan.dbVgInfoMap, name->dbname, strlen(name->dbname));

×

2140

  QRY_PARAM_CHECK(dbVgInfo);

×

2142

  if (find == NULL) {

×

2143

    output = taosMemoryMalloc(sizeof(SUseDbOutput));

×

2144

    code = buildDbVgInfoMap(pOperator, pMsgCb, name, pTaskInfo, output);

×

2145

    QUERY_CHECK_CODE(code, line, _return);

×

2146

    code = taosHashPut(pInfo->vtbScan.dbVgInfoMap, name->dbname, strlen(name->dbname), &output, POINTER_BYTES);

×

2147

    QUERY_CHECK_CODE(code, line, _return);

×

2149

    output = *find;

×

2152

  *dbVgInfo = output->dbVgroup;

×

2153

  return code;

×

2154

_return:

×

2155

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

×

2156

  freeUseDbOutput(output);

×

2157

  return code;

×

2160

int32_t extractColRefName(const char *colref, char **refDb, char** refTb, char** refCol) {

×

2161

  int32_t     code = TSDB_CODE_SUCCESS;

×

2162

  int32_t     line = 0;

×

2164

  const char *first_dot = strchr(colref, '.');

×

2165

  QUERY_CHECK_NULL(first_dot, code, line, _return, terrno)

×

2167

  const char *second_dot = strchr(first_dot + 1, '.');

×

2168

  QUERY_CHECK_NULL(second_dot, code, line, _return, terrno)

×

2170

  size_t db_len = first_dot - colref;

×

2171

  size_t table_len = second_dot - first_dot - 1;

×

2172

  size_t col_len = strlen(second_dot + 1);

×

2174

  *refDb = taosMemoryMalloc(db_len + 1);

×

2175

  *refTb = taosMemoryMalloc(table_len + 1);

×

2176

  *refCol = taosMemoryMalloc(col_len + 1);

×

2177

  QUERY_CHECK_NULL(*refDb, code, line, _return, terrno)

×

2178

  QUERY_CHECK_NULL(*refTb, code, line, _return, terrno)

×

2179

  QUERY_CHECK_NULL(*refCol, code, line, _return, terrno)

×

2181

  tstrncpy(*refDb, colref, db_len + 1);

×

2182

  tstrncpy(*refTb, first_dot + 1, table_len + 1);

×

2183

  tstrncpy(*refCol, second_dot + 1, col_len + 1);

×

2185

  return TSDB_CODE_SUCCESS;

×

2186

_return:

×

2187

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

×

2188

  if (*refDb) {

×

2189

    taosMemoryFree(*refDb);

×

2190

    *refDb = NULL;

×

2192

  if (*refTb) {

×

2193

    taosMemoryFree(*refTb);

×

2194

    *refTb = NULL;

×

2196

  if (*refCol) {

×

2197

    taosMemoryFree(*refCol);

×

2198

    *refCol = NULL;

×

2200

  return code;

×

2203

bool tableInfoNeedCollect(char *dbName, char *tbName, char *expectDbName, char *expectTbName) {

×

2204

  if (strncmp(varDataVal(tbName), expectTbName, varDataLen(tbName)) == 0 &&

×

2205

      strlen(expectTbName) == varDataLen(tbName) &&

×

2206

      strncmp(varDataVal(dbName), expectDbName, varDataLen(dbName)) == 0 &&

×

2207

      strlen(expectDbName) == varDataLen(dbName)) {

×

2208

    return true;

×

2210

  return false;

×

2213

int32_t getColRefInfo(SColRefInfo *pInfo, SArray* pDataBlock, int32_t index) {

×

2214

  int32_t          code = TSDB_CODE_SUCCESS;

×

2215

  int32_t          line = 0;

×

2217

  SColumnInfoData *pColNameCol = taosArrayGet(pDataBlock, 3);

×

2218

  SColumnInfoData *pUidCol = taosArrayGet(pDataBlock, 4);

×

2219

  SColumnInfoData *pColIdCol = taosArrayGet(pDataBlock, 5);

×

2220

  SColumnInfoData *pRefCol = taosArrayGet(pDataBlock, 6);

×

2221

  SColumnInfoData *pVgIdCol = taosArrayGet(pDataBlock, 7);

×

2222

  SColumnInfoData *pRefVerCol = taosArrayGet(pDataBlock, 8);

×

2224

  QUERY_CHECK_NULL(pColNameCol, code, line, _return, terrno)

×

2225

  QUERY_CHECK_NULL(pUidCol, code, line, _return, terrno)

×

2226

  QUERY_CHECK_NULL(pColIdCol, code, line, _return, terrno)

×

2227

  QUERY_CHECK_NULL(pRefCol, code, line, _return, terrno)

×

2228

  QUERY_CHECK_NULL(pVgIdCol, code, line, _return, terrno)

×

2229

  QUERY_CHECK_NULL(pRefVerCol, code, line, _return, terrno)

×

2231

  if (colDataIsNull_s(pRefCol, index)) {

×

2232

    pInfo->colrefName = NULL;

×

2234

    pInfo->colrefName = taosMemoryCalloc(varDataTLen(colDataGetData(pRefCol, index)), 1);

×

2235

    QUERY_CHECK_NULL(pInfo->colrefName, code, line, _return, terrno)

×

2236

    memcpy(pInfo->colrefName, varDataVal(colDataGetData(pRefCol, index)), varDataLen(colDataGetData(pRefCol, index)));

×

2237

    pInfo->colrefName[varDataLen(colDataGetData(pRefCol, index))] = 0;

×

2240

  pInfo->colName = taosMemoryCalloc(varDataTLen(colDataGetData(pColNameCol, index)), 1);

×

2241

  QUERY_CHECK_NULL(pInfo->colName, code, line, _return, terrno)

×

2242

  memcpy(pInfo->colName, varDataVal(colDataGetData(pColNameCol, index)), varDataLen(colDataGetData(pColNameCol, index)));

×

2243

  pInfo->colName[varDataLen(colDataGetData(pColNameCol, index))] = 0;

×

2245

  if (!colDataIsNull_s(pUidCol, index)) {

×

2246

    GET_TYPED_DATA(pInfo->uid, int64_t, TSDB_DATA_TYPE_BIGINT, colDataGetNumData(pUidCol, index), 0);

×

2248

  if (!colDataIsNull_s(pColIdCol, index)) {

×

2249

    GET_TYPED_DATA(pInfo->colId, int32_t, TSDB_DATA_TYPE_INT, colDataGetNumData(pColIdCol, index), 0);

×

2251

  if (!colDataIsNull_s(pVgIdCol, index)) {

×

2252

    GET_TYPED_DATA(pInfo->vgId, int32_t, TSDB_DATA_TYPE_INT, colDataGetNumData(pVgIdCol, index), 0);

×

2255

_return:

×

2256

  return code;

×

2259

int32_t processOrgTbVg(SVtbScanDynCtrlInfo* pVtbScan, SExecTaskInfo* pTaskInfo, int32_t rversion) {

×

2260

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2261

  int32_t                    line = 0;

×

2263

  if (pTaskInfo->pStreamRuntimeInfo == NULL) {

×

2264

    return code;

×

2267

  if (pVtbScan->existOrgTbVg == NULL) {

×

2268

    pVtbScan->existOrgTbVg = pVtbScan->curOrgTbVg;

×

2269

    pVtbScan->curOrgTbVg = NULL;

×

2272

  if (pVtbScan->curOrgTbVg != NULL) {

×

2274

    void*   pCurIter = NULL;

×

2275

    SArray* tmpArray = NULL;

×

2276

    while ((pCurIter = taosHashIterate(pVtbScan->curOrgTbVg, pCurIter))) {

×

2277

      int32_t* vgId = (int32_t*)taosHashGetKey(pCurIter, NULL);

×

2278

      if (taosHashGet(pVtbScan->existOrgTbVg, vgId, sizeof(int32_t)) == NULL) {

×

2279

        if (tmpArray == NULL) {

×

2280

          tmpArray = taosArrayInit(1, sizeof(int32_t));

×

2281

          QUERY_CHECK_NULL(tmpArray, code, line, _return, terrno)

×

2283

        QUERY_CHECK_NULL(taosArrayPush(tmpArray, vgId), code, line, _return, terrno)

×

2286

    if (tmpArray == NULL) {

×

2287

      return TSDB_CODE_SUCCESS;

×

2289

    if (tmpArray != NULL && pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.addVgIds == NULL) {

×

2290

      SArray* expiredInfo = atomic_load_ptr(&pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.addedVgInfo);

×

2291

      if (expiredInfo && expiredInfo == atomic_val_compare_exchange_ptr(&pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.addedVgInfo, expiredInfo, NULL)) {

×

2292

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

×

2293

          SStreamTaskAddr* vgInfo = (SStreamTaskAddr*)taosArrayGet(expiredInfo, i);

×

2294

          QUERY_CHECK_NULL(taosArrayPush(tmpArray, &vgInfo->nodeId), code, line, _return, terrno)

×

2296

        taosArrayDestroy(expiredInfo);

×

2298

      if (atomic_val_compare_exchange_ptr(&pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.addVgIds, NULL, tmpArray)) {

×

2299

        taosArrayDestroy(tmpArray);

×

2302

    atomic_store_64(&pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.uid, (int64_t)(pVtbScan->isSuperTable ? pVtbScan->suid : pVtbScan->uid));

×

2303

    (void)atomic_val_compare_exchange_8(pTaskInfo->pStreamRuntimeInfo->vtableDeployGot, 0, 1);

×

2304

    taosHashClear(pVtbScan->curOrgTbVg);

×

2305

    pVtbScan->needRedeploy = true;

×

2306

    pVtbScan->rversion = rversion;

×

2307

    return TSDB_CODE_STREAM_VTABLE_NEED_REDEPLOY;

×

2309

  return code;

×

2310

_return:

×

2311

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

×

2312

  return code;

×

2315

int32_t getVgIdFromColref(SOperatorInfo* pOperator, const char* colRef, int32_t* vgId) {

×

2316

  int32_t                    code =TSDB_CODE_SUCCESS;

×

2317

  int32_t                    line = 0;

×

2318

  char*                      refDbName = NULL;

×

2319

  char*                      refTbName = NULL;

×

2320

  char*                      refColName = NULL;

×

2321

  SDBVgInfo*                 dbVgInfo = NULL;

×

2322

  SName                      name = {0};

×

2323

  char                       dbFname[TSDB_DB_FNAME_LEN] = {0};

×

2324

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2326

  code = extractColRefName(colRef, &refDbName, &refTbName, &refColName);

×

2327

  QUERY_CHECK_CODE(code, line, _return);

×

2329

  toName(pInfo->vtbScan.acctId, refDbName, refTbName, &name);

×

2331

  code = getDbVgInfo(pOperator, &name, &dbVgInfo);

×

2332

  QUERY_CHECK_CODE(code, line, _return);

×

2334

  code = tNameGetFullDbName(&name, dbFname);

×

2335

  QUERY_CHECK_CODE(code, line, _return);

×

2337

  code = getVgId(dbVgInfo, dbFname, vgId, name.tname);

×

2338

  QUERY_CHECK_CODE(code, line, _return);

×

2340

_return:

×

2341

  if (code) {

×

2342

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

×

2344

  taosMemoryFree(refDbName);

×

2345

  taosMemoryFree(refTbName);

×

2346

  taosMemoryFree(refColName);

×

2347

  return code;

×

2350

static int32_t generateTagArrayByTagBlockAndSave(SHashObj* vtbUidTagListMap, tb_uid_t uid, SSDataBlock *pTagVal, int32_t rowIdx) {

×

2351

  int32_t code = TSDB_CODE_SUCCESS;

×

2352

  int32_t line = 0;

×

2353

  STagVal tagVal = {0};

×

2356

  SArray* pTagList = taosArrayInit(1, sizeof(STagVal));

×

2357

  QUERY_CHECK_NULL(pTagList, code, line, _return, terrno)

×

2359

  for (int32_t k = 0; k < taosArrayGetSize(pTagVal->pDataBlock) - 1; k++) {

×

2360

    SColumnInfoData *pTagCol = taosArrayGet(pTagVal->pDataBlock, k);

×

2361

    QUERY_CHECK_NULL(pTagCol, code, line, _return, terrno)

×

2362

    tagVal.type = pTagCol->info.type;

×

2363

    tagVal.cid = pTagCol->info.colId;

×

2364

    if (!colDataIsNull_s(pTagCol, rowIdx)) {

×

2365

      char*   pData = colDataGetData(pTagCol, rowIdx);

×

2366

      if (IS_VAR_DATA_TYPE(pTagCol->info.type)) {

×

2367

        tagVal.nData = varDataLen(pData);

×

2368

        tagVal.pData = taosMemoryMalloc(tagVal.nData);

×

2369

        QUERY_CHECK_NULL(tagVal.pData, code, line, _return, terrno)

×

2370

        memcpy(tagVal.pData, varDataVal(pData), varDataLen(pData));

×

2371

        QUERY_CHECK_NULL(taosArrayPush(pTagList, &tagVal), code, line, _return, terrno)

×

2373

        memcpy(&tagVal.i64, pData, tDataTypes[pTagCol->info.type].bytes);

×

2374

        QUERY_CHECK_NULL(taosArrayPush(pTagList, &tagVal), code, line, _return, terrno)

×

2377

      tagVal.pData = NULL;

×

2378

      tagVal.nData = 0;

×

2379

      QUERY_CHECK_NULL(taosArrayPush(pTagList, &tagVal), code, line, _return, terrno)

×

2381

    tagVal = (STagVal){0};

×

2383

  code = taosHashPut(vtbUidTagListMap, &uid, sizeof(uid), &pTagList, POINTER_BYTES);

×

2384

  QUERY_CHECK_CODE(code, line, _return);

×

2386

  return code;

×

2387

_return:

×

2388

  if (tagVal.pData) {

×

2389

    taosMemoryFreeClear(tagVal.pData);

×

2391

  if (pTagList) {

×

2392

    taosArrayDestroyEx(pTagList, destroyTagVal);

×

2394

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

×

2395

  return code;

×

2398

int32_t virtualTableScanProcessColRefInfo(SOperatorInfo* pOperator, SArray* pColRefInfo, tb_uid_t* uid, int32_t* vgId,

×

2400

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2401

  int32_t                    line = 0;

×

2402

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2403

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2404

  SDBVgInfo*                 dbVgInfo = NULL;

×

2405

  SHashObj*                  refMap = NULL;

×

2407

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

×

2408

    SColRefInfo *pKV = (SColRefInfo*)taosArrayGet(pColRefInfo, j);

×

2409

    *uid = pKV->uid;

×

2410

    *vgId = pKV->vgId;

×

2411

    if (pKV->colrefName != NULL && colNeedScan(pOperator, pKV->colId)) {

×

2412

      char*   refDbName = NULL;

×

2413

      char*   refTbName = NULL;

×

2414

      char*   refColName = NULL;

×

2415

      SName   name = {0};

×

2416

      char    dbFname[TSDB_DB_FNAME_LEN] = {0};

×

2417

      char    orgTbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2419

      if (ppRefMap != NULL) {

×

2421

        if (refMap == NULL) {

×

2422

          refMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK);

×

2423

          QUERY_CHECK_NULL(refMap, code, line, _return, terrno)

×

2425

        code = addRefColIdToRefMap(refMap, pKV->colrefName, pKV->colId);

×

2426

        QUERY_CHECK_CODE(code, line, _return);

×

2430

      code = extractColRefName(pKV->colrefName, &refDbName, &refTbName, &refColName);

×

2431

      QUERY_CHECK_CODE(code, line, _return);

×

2433

      toName(pInfo->vtbScan.acctId, refDbName, refTbName, &name);

×

2435

      code = getDbVgInfo(pOperator, &name, &dbVgInfo);

×

2436

      QUERY_CHECK_CODE(code, line, _return);

×

2437

      code = tNameGetFullDbName(&name, dbFname);

×

2438

      QUERY_CHECK_CODE(code, line, _return);

×

2439

      code = tNameGetFullTableName(&name, orgTbFName);

×

2440

      QUERY_CHECK_CODE(code, line, _return);

×

2442

      void *pVal = taosHashGet(pVtbScan->otbNameToOtbInfoMap, orgTbFName, sizeof(orgTbFName));

×

2443

      if (!pVal) {

×

2444

        SOrgTbInfo orgTbInfo = {0};

×

2445

        code = getVgId(dbVgInfo, dbFname, &orgTbInfo.vgId, name.tname);

×

2446

        QUERY_CHECK_CODE(code, line, _return);

×

2447

        tstrncpy(orgTbInfo.tbName, orgTbFName, sizeof(orgTbInfo.tbName));

×

2448

        orgTbInfo.colMap = taosArrayInit(10, sizeof(SColIdNameKV));

×

2449

        QUERY_CHECK_NULL(orgTbInfo.colMap, code, line, _return, terrno)

×

2450

        SColIdNameKV colIdNameKV = {0};

×

2451

        colIdNameKV.colId = pKV->colId;

×

2452

        tstrncpy(colIdNameKV.colName, refColName, sizeof(colIdNameKV.colName));

×

2453

        QUERY_CHECK_NULL(taosArrayPush(orgTbInfo.colMap, &colIdNameKV), code, line, _return, terrno)

×

2454

        code = taosHashPut(pVtbScan->otbNameToOtbInfoMap, orgTbFName, sizeof(orgTbFName), &orgTbInfo, sizeof(orgTbInfo));

×

2455

        QUERY_CHECK_CODE(code, line, _return);

×

2457

        SOrgTbInfo *tbInfo = (SOrgTbInfo *)pVal;

×

2458

        SColIdNameKV colIdNameKV = {0};

×

2459

        colIdNameKV.colId = pKV->colId;

×

2460

        tstrncpy(colIdNameKV.colName, refColName, sizeof(colIdNameKV.colName));

×

2461

        QUERY_CHECK_NULL(taosArrayPush(tbInfo->colMap, &colIdNameKV), code, line, _return, terrno)

×

2463

      taosMemoryFree(refDbName);

×

2464

      taosMemoryFree(refTbName);

×

2465

      taosMemoryFree(refColName);

×

2469

  if (ppRefMap != NULL) {

×

2470

    *ppRefMap = refMap;

×

2473

  return code;

×

2475

_return:

×

2476

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

×

2477

  if (refMap) {

×

2478

    taosHashCleanup(refMap);

×

2480

  return code;

×

2483

static int32_t getTagBlockAndProcess(SOperatorInfo* pOperator, bool hasPartition) {

×

2484

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2485

  int32_t                    line = 0;

×

2486

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2487

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2488

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2489

  SArray*                    pColRefArray = NULL;

×

2490

  SOperatorInfo*             pSystableScanOp = pOperator->pDownstream[0];

×

2491

  SOperatorInfo*             pTagScanOp = pOperator->pDownstream[1];

×

2493

  pVtbScan->vtbUidTagListMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

2494

  QUERY_CHECK_NULL(pVtbScan->vtbUidTagListMap, code, line, _return, terrno)

×

2495

  taosHashSetFreeFp(pVtbScan->vtbUidTagListMap, destroyTagList);

×

2496

  if (hasPartition) {

×

2497

    pVtbScan->vtbUidToGroupIdMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

2498

    pVtbScan->vtbGroupIdTagListMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);

×

2499

    QUERY_CHECK_NULL(pVtbScan->vtbUidToGroupIdMap, code, line, _return, terrno)

×

2500

    QUERY_CHECK_NULL(pVtbScan->vtbGroupIdTagListMap, code, line, _return, terrno)

×

2501

    taosHashSetFreeFp(pVtbScan->vtbGroupIdTagListMap, destroyVtbUidTagListMap);

×

2504

  while (true) {

×

2505

    SSDataBlock *pTagVal = NULL;

×

2506

    code = pTagScanOp->fpSet.getNextFn(pTagScanOp, &pTagVal);

×

2507

    QUERY_CHECK_CODE(code, line, _return);

×

2508

    if (pTagVal == NULL) {

×

2509

      break;

×

2511

    SHashObj *vtbUidTagListMap = NULL;

×

2512

    if (hasPartition) {

×

2513

      void* pIter = taosHashGet(pVtbScan->vtbGroupIdTagListMap, &pTagVal->info.id.groupId, sizeof(pTagVal->info.id.groupId));

×

2514

      if (pIter) {

×

2515

        vtbUidTagListMap = *(SHashObj**)pIter;

×

2517

        vtbUidTagListMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

2518

        QUERY_CHECK_NULL(vtbUidTagListMap, code, line, _return, terrno)

×

2519

        taosHashSetFreeFp(vtbUidTagListMap, destroyTagList);

×

2521

        code = taosHashPut(pVtbScan->vtbGroupIdTagListMap, &pTagVal->info.id.groupId, sizeof(pTagVal->info.id.groupId), &vtbUidTagListMap, POINTER_BYTES);

×

2522

        QUERY_CHECK_CODE(code, line, _return);

×

2525

      vtbUidTagListMap = pVtbScan->vtbUidTagListMap;

×

2528

    SColumnInfoData *pUidCol = taosArrayGetLast(pTagVal->pDataBlock);

×

2529

    QUERY_CHECK_NULL(pUidCol, code, line, _return, terrno)

×

2530

    for (int32_t i = 0; i < pTagVal->info.rows; i++) {

×

2531

      tb_uid_t uid = 0;

×

2532

      if (!colDataIsNull_s(pUidCol, i)) {

×

2533

        GET_TYPED_DATA(uid, int64_t, TSDB_DATA_TYPE_BIGINT, colDataGetNumData(pUidCol, i), 0);

×

2534

        QUERY_CHECK_CODE(code, line, _return);

×

2537

      code = generateTagArrayByTagBlockAndSave(vtbUidTagListMap, uid, pTagVal, i);

×

2538

      QUERY_CHECK_CODE(code, line, _return);

×

2540

      if (hasPartition) {

×

2541

        code = taosHashPut(pVtbScan->vtbUidToGroupIdMap, &uid, sizeof(uid), &pTagVal->info.id.groupId, sizeof(pTagVal->info.id.groupId));

×

2542

        QUERY_CHECK_CODE(code, line, _return);

×

2547

  return code;

×

2549

_return:

×

2550

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

×

2551

  return code;

×

2554

static int32_t processChildTableListAndGenerateOrgTbInfoMap(SOperatorInfo* pOperator) {

×

2555

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2556

  int32_t                    line = 0;

×

2557

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2558

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2559

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2560

  SArray*                    pColRefArray = NULL;

×

2561

  SOperatorInfo*             pSystableScanOp = pOperator->pDownstream[0];

×

2562

  SOperatorInfo*             pTagScanOp = pOperator->pDownstream[1];

×

2564

  pVtbScan->vtbUidToVgIdMapMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

2565

  QUERY_CHECK_NULL(pVtbScan->vtbUidToVgIdMapMap, code, line, _return, terrno)

×

2567

  for (int32_t i = 0; i < taosArrayGetSize(pVtbScan->childTableList); i++) {

×

2568

    SHashObj* otbVgIdToOtbInfoArrayMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);

×

2569

    QUERY_CHECK_NULL(otbVgIdToOtbInfoArrayMap, code, line, _return, terrno)

×

2571

    SArray* pColRefInfo = (SArray*)taosArrayGetP(pVtbScan->childTableList, i);

×

2572

    QUERY_CHECK_NULL(pColRefInfo, code, line, _return, terrno)

×

2574

    tb_uid_t uid = 0;

×

2575

    int32_t  vgId = 0;

×

2576

    code = virtualTableScanProcessColRefInfo(pOperator, pColRefInfo, &uid, &vgId, NULL);

×

2577

    QUERY_CHECK_CODE(code, line, _return);

×

2579

    size_t len = 0;

×

2580

    void*  pOrgTbInfo = taosHashIterate(pVtbScan->otbNameToOtbInfoMap, NULL);

×

2581

    while (pOrgTbInfo != NULL) {

×

2582

      char*       key = taosHashGetKey(pOrgTbInfo, &len);

×

2583

      SOrgTbInfo* orgTbInfo = (SOrgTbInfo*)pOrgTbInfo;

×

2585

      void* pIter = taosHashGet(otbVgIdToOtbInfoArrayMap, &orgTbInfo->vgId, sizeof(orgTbInfo->vgId));

×

2586

      if (!pIter) {

×

2587

        SArray* pOrgTbInfoArray = taosArrayInit(1, sizeof(SOrgTbInfo));

×

2588

        QUERY_CHECK_NULL(pOrgTbInfoArray, code, line, _return, terrno)

×

2589

        QUERY_CHECK_NULL(taosArrayPush(pOrgTbInfoArray, orgTbInfo), code, line, _return, terrno)

×

2590

        code = taosHashPut(otbVgIdToOtbInfoArrayMap, &orgTbInfo->vgId, sizeof(orgTbInfo->vgId), &pOrgTbInfoArray, POINTER_BYTES);

×

2591

        QUERY_CHECK_CODE(code, line, _return);

×

2593

        SArray* pOrgTbInfoArray = *(SArray**)pIter;

×

2594

        QUERY_CHECK_NULL(pOrgTbInfoArray, code, line, _return, terrno)

×

2595

        QUERY_CHECK_NULL(taosArrayPush(pOrgTbInfoArray, orgTbInfo), code, line, _return, terrno)

×

2598

      pOrgTbInfo = taosHashIterate(pVtbScan->otbNameToOtbInfoMap, pOrgTbInfo);

×

2600

      code = taosHashRemove(pVtbScan->otbNameToOtbInfoMap, key, len);

×

2601

      QUERY_CHECK_CODE(code, line, _return);

×

2604

    code = taosHashPut(pVtbScan->vtbUidToVgIdMapMap, &uid, sizeof(uid), &otbVgIdToOtbInfoArrayMap, POINTER_BYTES);

×

2605

    QUERY_CHECK_CODE(code, line, _return);

×

2608

  return code;

×

2609

_return:

×

2610

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

×

2611

  return code;

×

2614

static int32_t buildOrgTbInfoSingle(SOperatorInfo* pOperator, bool hasPartition) {

×

2615

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2616

  int32_t                    line = 0;

×

2618

  code = processChildTableListAndGenerateOrgTbInfoMap(pOperator);

×

2619

  QUERY_CHECK_CODE(code, line, _return);

×

2622

  code = getTagBlockAndProcess(pOperator, hasPartition);

×

2623

  QUERY_CHECK_CODE(code, line, _return);

×

2625

  return code;

×

2626

_return:

×

2627

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

×

2628

  return code;

×

2631

static int32_t buildOrgTbInfoBatch(SOperatorInfo* pOperator, bool hasPartition) {

×

2632

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2633

  int32_t                    line = 0;

×

2634

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2635

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2636

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2637

  SArray*                    pColRefArray = NULL;

×

2638

  SOperatorInfo*             pSystableScanOp = pOperator->pDownstream[0];

×

2639

  SOperatorInfo*             pTagScanOp = pOperator->pDownstream[1];

×

2641

  if (hasPartition) {

×

2642

    pVtbScan->vtbUidToGroupIdMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);

×

2643

    pVtbScan->vtbGroupIdTagListMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);

×

2644

    pVtbScan->vtbGroupIdToVgIdMapMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_ENTRY_LOCK);

×

2646

    QUERY_CHECK_NULL(pVtbScan->vtbGroupIdToVgIdMapMap, code, line, _return, terrno)

×

2647

    QUERY_CHECK_NULL(pVtbScan->vtbUidToGroupIdMap, code, line, _return, terrno)

×

2648

    QUERY_CHECK_NULL(pVtbScan->vtbGroupIdTagListMap, code, line, _return, terrno)

×

2649

    taosHashSetFreeFp(pVtbScan->vtbGroupIdToVgIdMapMap, destroyOtbVgIdToOtbInfoArrayMap);

×

2651

    pVtbScan->otbVgIdToOtbInfoArrayMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);

×

2652

    QUERY_CHECK_NULL(pVtbScan->otbVgIdToOtbInfoArrayMap, code, line, _return, terrno)

×

2655

  while (true && hasPartition) {

×

2656

    SSDataBlock* pTagVal = NULL;

×

2657

    code = pTagScanOp->fpSet.getNextFn(pTagScanOp, &pTagVal);

×

2658

    QUERY_CHECK_CODE(code, line, _return);

×

2659

    if (pTagVal == NULL) {

×

2660

      break;

×

2663

    SColumnInfoData *pUidCol = taosArrayGetLast(pTagVal->pDataBlock);

×

2664

    QUERY_CHECK_NULL(pUidCol, code, line, _return, terrno)

×

2665

    for (int32_t i = 0; i < pTagVal->info.rows; i++) {

×

2666

      tb_uid_t uid = 0;

×

2667

      if (!colDataIsNull_s(pUidCol, i)) {

×

2668

        GET_TYPED_DATA(uid, int64_t, TSDB_DATA_TYPE_BIGINT, colDataGetNumData(pUidCol, i), 0);

×

2669

        QUERY_CHECK_CODE(code, line, _return);

×

2671

      code = taosHashPut(pVtbScan->vtbUidToGroupIdMap, &uid, sizeof(uid), &pTagVal->info.id.groupId, sizeof(pTagVal->info.id.groupId));

×

2672

      QUERY_CHECK_CODE(code, line, _return);

×

2674

    code = taosHashPut(pVtbScan->vtbGroupIdTagListMap, &pTagVal->info.id.groupId, sizeof(pTagVal->info.id.groupId), NULL, 0);

×

2675

    QUERY_CHECK_CODE(code, line, _return);

×

2678

  for (int32_t i = 0; i < taosArrayGetSize(pVtbScan->childTableList); i++) {

×

2679

    SArray* pColRefInfo = (SArray*)taosArrayGetP(pVtbScan->childTableList, i);

×

2680

    QUERY_CHECK_NULL(pColRefInfo, code, line, _return, terrno)

×

2681

    tb_uid_t uid = 0;

×

2682

    int32_t  vgId = 0;

×

2683

    code = virtualTableScanProcessColRefInfo(pOperator, pColRefInfo, &uid, &vgId, NULL);

×

2684

    QUERY_CHECK_CODE(code, line, _return);

×

2686

    SHashObj* otbVgIdToOtbInfoArrayMap = NULL;

×

2687

    if (hasPartition) {

×

2688

      uint64_t* groupId = (uint64_t *)taosHashGet(pVtbScan->vtbUidToGroupIdMap, &uid, sizeof(uid));

×

2689

      QUERY_CHECK_NULL(groupId, code, line, _return, terrno)

×

2691

      void* pHashIter = taosHashGet(pVtbScan->vtbGroupIdToVgIdMapMap, groupId, sizeof(*groupId));

×

2692

      if (pHashIter) {

×

2693

        otbVgIdToOtbInfoArrayMap = *(SHashObj**)pHashIter;

×

2695

        otbVgIdToOtbInfoArrayMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);

×

2696

        QUERY_CHECK_NULL(otbVgIdToOtbInfoArrayMap, code, line, _return, terrno)

×

2697

        code = taosHashPut(pVtbScan->vtbGroupIdToVgIdMapMap, groupId, sizeof(*groupId), &otbVgIdToOtbInfoArrayMap, POINTER_BYTES);

×

2698

        QUERY_CHECK_CODE(code, line, _return);

×

2701

      otbVgIdToOtbInfoArrayMap = pVtbScan->otbVgIdToOtbInfoArrayMap;

×

2704

    size_t len = 0;

×

2705

    void*  pOrgTbInfo = taosHashIterate(pVtbScan->otbNameToOtbInfoMap, NULL);

×

2706

    while (pOrgTbInfo != NULL) {

×

2707

      char*       key = taosHashGetKey(pOrgTbInfo, &len);

×

2708

      SOrgTbInfo* orgTbInfo = (SOrgTbInfo*)pOrgTbInfo;

×

2709

      void* pIter = taosHashGet(otbVgIdToOtbInfoArrayMap, &orgTbInfo->vgId, sizeof(orgTbInfo->vgId));

×

2710

      if (!pIter) {

×

2711

        SArray* pOrgTbInfoArray = taosArrayInit(1, sizeof(SOrgTbInfo));

×

2712

        QUERY_CHECK_NULL(pOrgTbInfoArray, code, line, _return, terrno)

×

2713

        QUERY_CHECK_NULL(taosArrayPush(pOrgTbInfoArray, orgTbInfo), code, line, _return, terrno)

×

2714

        code = taosHashPut(otbVgIdToOtbInfoArrayMap, &orgTbInfo->vgId, sizeof(orgTbInfo->vgId), &pOrgTbInfoArray, POINTER_BYTES);

×

2715

        QUERY_CHECK_CODE(code, line, _return);

×

2717

        SArray* pOrgTbInfoArray = *(SArray**)pIter;

×

2718

        QUERY_CHECK_NULL(pOrgTbInfoArray, code, line, _return, terrno)

×

2719

        QUERY_CHECK_NULL(taosArrayPush(pOrgTbInfoArray, orgTbInfo), code, line, _return, terrno)

×

2722

      pOrgTbInfo = taosHashIterate(pVtbScan->otbNameToOtbInfoMap, pOrgTbInfo);

×

2724

      code = taosHashRemove(pVtbScan->otbNameToOtbInfoMap, key, len);

×

2725

      QUERY_CHECK_CODE(code, line, _return);

×

2728

  return code;

×

2729

_return:

×

2730

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

×

2731

  return code;

×

2734

int32_t buildVirtualSuperTableScanChildTableMap(SOperatorInfo* pOperator) {

×

2735

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2736

  int32_t                    line = 0;

×

2737

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2738

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2739

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2740

  SArray*                    pColRefArray = NULL;

×

2741

  SOperatorInfo*             pSystableScanOp = NULL;

×

2743

  pVtbScan->childTableMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

2744

  QUERY_CHECK_NULL(pVtbScan->childTableMap, code, line, _return, terrno)

×

2746

  if (pInfo->qType == DYN_QTYPE_VTB_AGG || pInfo->qType == DYN_QTYPE_VTB_INTERVAL) {

×

2747

    pVtbScan->otbNameToOtbInfoMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

2748

    QUERY_CHECK_NULL(pVtbScan->otbNameToOtbInfoMap, code, line, _return, terrno)

×

2749

    pSystableScanOp = pOperator->pDownstream[0];

×

2750

  } else if (pInfo->qType == DYN_QTYPE_VTB_WINDOW || pInfo->qType == DYN_QTYPE_VTB_TS_SCAN) {

×

2751

    pVtbScan->otbNameToOtbInfoMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

2752

    QUERY_CHECK_NULL(pVtbScan->otbNameToOtbInfoMap, code, line, _return, terrno)

×

2753

    pSystableScanOp = pOperator->pDownstream[1];

×

2755

    pSystableScanOp = pOperator->pDownstream[1];

×

2758

  while (true) {

×

2759

    SSDataBlock *pChildInfo = NULL;

×

2760

    code = pSystableScanOp->fpSet.getNextFn(pSystableScanOp, &pChildInfo);

×

2761

    QUERY_CHECK_CODE(code, line, _return);

×

2762

    if (pChildInfo == NULL) {

×

2763

      break;

×

2765

    SColumnInfoData *pTableNameCol = taosArrayGet(pChildInfo->pDataBlock, 0);

×

2766

    SColumnInfoData *pStbNameCol = taosArrayGet(pChildInfo->pDataBlock, 1);

×

2767

    SColumnInfoData *pDbNameCol = taosArrayGet(pChildInfo->pDataBlock, 2);

×

2769

    QUERY_CHECK_NULL(pTableNameCol, code, line, _return, terrno)

×

2770

    QUERY_CHECK_NULL(pStbNameCol, code, line, _return, terrno)

×

2771

    QUERY_CHECK_NULL(pDbNameCol, code, line, _return, terrno)

×

2773

    for (int32_t i = 0; i < pChildInfo->info.rows; i++) {

×

2774

      if (!colDataIsNull_s(pStbNameCol, i)) {

×

2775

        char* stbrawname = colDataGetData(pStbNameCol, i);

×

2776

        char* dbrawname = colDataGetData(pDbNameCol, i);

×

2777

        char *ctbName = colDataGetData(pTableNameCol, i);

×

2779

        if (tableInfoNeedCollect(dbrawname, stbrawname, pInfo->vtbScan.dbName, pInfo->vtbScan.tbName)) {

×

2780

          SColRefInfo info = {0};

×

2781

          code = getColRefInfo(&info, pChildInfo->pDataBlock, i);

×

2782

          QUERY_CHECK_CODE(code, line, _return);

×

2784

          if (pInfo->qType == DYN_QTYPE_VTB_SCAN) {

×

2785

            if (pInfo->vtbScan.dynTbUid != 0 && info.uid != pInfo->vtbScan.dynTbUid) {

×

2786

              qTrace("dynQueryCtrl tb uid filter, info uid:%" PRIu64 ", dyn tb uid:%" PRIu64, info.uid,

×

2788

              destroyColRefInfo(&info);

×

2789

              continue;

×

2792

            if (pTaskInfo->pStreamRuntimeInfo) {

×

2793

              if (pVtbScan->curOrgTbVg == NULL) {

×

2794

                pVtbScan->curOrgTbVg = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

2795

                QUERY_CHECK_NULL(pVtbScan->curOrgTbVg, code, line, _return, terrno)

×

2798

              if (info.colrefName) {

×

2800

                code = getVgIdFromColref(pOperator, info.colrefName, &vgId);

×

2801

                QUERY_CHECK_CODE(code, line, _return);

×

2802

                code = taosHashPut(pVtbScan->curOrgTbVg, &vgId, sizeof(vgId), NULL, 0);

×

2803

                QUERY_CHECK_CODE(code, line, _return);

×

2808

          if (taosHashGet(pVtbScan->childTableMap, varDataVal(ctbName), varDataLen(ctbName)) == NULL) {

×

2809

            pColRefArray = taosArrayInit(1, sizeof(SColRefInfo));

×

2810

            QUERY_CHECK_NULL(pColRefArray, code, line, _return, terrno)

×

2811

            QUERY_CHECK_NULL(taosArrayPush(pColRefArray, &info), code, line, _return, terrno)

×

2812

            int32_t tableIdx = (int32_t)taosArrayGetSize(pVtbScan->childTableList);

×

2813

            QUERY_CHECK_NULL(taosArrayPush(pVtbScan->childTableList, &pColRefArray), code, line, _return, terrno)

×

2814

            code = taosHashPut(pVtbScan->childTableMap, varDataVal(ctbName), varDataLen(ctbName), &tableIdx, sizeof(tableIdx));

×

2815

            QUERY_CHECK_CODE(code, line, _return);

×

2817

            int32_t *tableIdx = (int32_t*)taosHashGet(pVtbScan->childTableMap, varDataVal(ctbName), varDataLen(ctbName));

×

2818

            QUERY_CHECK_NULL(tableIdx, code, line, _return, terrno)

×

2819

            pColRefArray = (SArray *)taosArrayGetP(pVtbScan->childTableList, *tableIdx);

×

2820

            QUERY_CHECK_NULL(pColRefArray, code, line, _return, terrno)

×

2821

            QUERY_CHECK_NULL(taosArrayPush(pColRefArray, &info), code, line, _return, terrno)

×

2828

  switch (pInfo->qType) {

×

2829

    case DYN_QTYPE_VTB_TS_SCAN:

×

2832

      code = buildOrgTbInfoBatch(pOperator, false);

×

2833

      break;

×

2835

    case DYN_QTYPE_VTB_AGG: {

×

2836

      if (pVtbScan->batchProcessChild) {

×

2837

        code = buildOrgTbInfoBatch(pOperator, pVtbScan->hasPartition);

×

2839

        code = buildOrgTbInfoSingle(pOperator, pVtbScan->hasPartition);

×

2841

      break;

×

2843

    case DYN_QTYPE_VTB_SCAN: {

×

2844

      code = processOrgTbVg(pVtbScan, pTaskInfo, 1);

×

2845

      break;

×

2847

    default: {

×

2848

      code = TSDB_CODE_PLAN_INVALID_DYN_CTRL_TYPE;

×

2849

      break;

×

2853

  QUERY_CHECK_CODE(code, line, _return);

×

2855

_return:

×

2856

  if (code) {

×

2857

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

×

2859

  return code;

×

2862

int32_t buildVirtualNormalChildTableScanChildTableMap(SOperatorInfo* pOperator) {

×

2863

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2864

  int32_t                    line = 0;

×

2865

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2866

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2867

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2868

  SArray*                    pColRefInfo = pInfo->vtbScan.colRefInfo;

×

2869

  SOperatorInfo*             pSystableScanOp = pOperator->pDownstream[1];

×

2870

  int32_t                    rversion = 0;

×

2872

  pInfo->vtbScan.colRefInfo = taosArrayInit(1, sizeof(SColRefInfo));

×

2873

  QUERY_CHECK_NULL(pInfo->vtbScan.colRefInfo, code, line, _return, terrno)

×

2875

  while (true) {

×

2876

    SSDataBlock *pTableInfo = NULL;

×

2877

    code = pSystableScanOp->fpSet.getNextFn(pSystableScanOp, &pTableInfo);

×

2878

    if (pTableInfo == NULL) {

×

2879

      break;

×

2882

    SColumnInfoData *pTableNameCol = taosArrayGet(pTableInfo->pDataBlock, 0);

×

2883

    SColumnInfoData *pDbNameCol = taosArrayGet(pTableInfo->pDataBlock, 2);

×

2884

    SColumnInfoData *pRefVerCol = taosArrayGet(pTableInfo->pDataBlock, 8);

×

2886

    QUERY_CHECK_NULL(pTableNameCol, code, line, _return, terrno)

×

2887

    QUERY_CHECK_NULL(pDbNameCol, code, line, _return, terrno)

×

2888

    QUERY_CHECK_NULL(pRefVerCol, code, line, _return, terrno)

×

2890

    for (int32_t i = 0; i < pTableInfo->info.rows; i++) {

×

2891

      if (!colDataIsNull_s(pRefVerCol, i)) {

×

2892

        GET_TYPED_DATA(rversion, int32_t, TSDB_DATA_TYPE_INT, colDataGetNumData(pRefVerCol, i), 0);

×

2895

      if (!colDataIsNull_s(pTableNameCol, i)) {

×

2896

        char* tbrawname = colDataGetData(pTableNameCol, i);

×

2897

        char* dbrawname = colDataGetData(pDbNameCol, i);

×

2898

        QUERY_CHECK_NULL(tbrawname, code, line, _return, terrno)

×

2899

        QUERY_CHECK_NULL(dbrawname, code, line, _return, terrno)

×

2901

        if (tableInfoNeedCollect(dbrawname, tbrawname, pInfo->vtbScan.dbName, pInfo->vtbScan.tbName)) {

×

2902

          SColRefInfo info = {0};

×

2903

          code = getColRefInfo(&info, pTableInfo->pDataBlock, i);

×

2904

          QUERY_CHECK_CODE(code, line, _return);

×

2906

          if ((rversion != pVtbScan->rversion || pVtbScan->existOrgTbVg == NULL) && info.colrefName) {

×

2907

            if (pVtbScan->curOrgTbVg == NULL) {

×

2908

              pVtbScan->curOrgTbVg = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

2909

              QUERY_CHECK_NULL(pVtbScan->curOrgTbVg, code, line, _return, terrno)

×

2912

            code = getVgIdFromColref(pOperator, info.colrefName, &vgId);

×

2913

            QUERY_CHECK_CODE(code, line, _return);

×

2914

            code = taosHashPut(pVtbScan->curOrgTbVg, &vgId, sizeof(vgId), NULL, 0);

×

2915

            QUERY_CHECK_CODE(code, line, _return);

×

2918

          QUERY_CHECK_NULL(taosArrayPush(pInfo->vtbScan.colRefInfo, &info), code, line, _return, terrno)

×

2923

  code = processOrgTbVg(pVtbScan, pTaskInfo, rversion);

×

2924

  QUERY_CHECK_CODE(code, line, _return);

×

2926

_return:

×

2927

  if (code) {

×

2928

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

×

2930

  return code;

×

2933

int32_t virtualTableScanCheckNeedRedeploy(SOperatorInfo* pOperator) {

×

2934

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2935

  int32_t                    line = 0;

×

2936

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2937

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2938

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2940

  SArray *tmpArray = NULL;

×

2941

  tmpArray = atomic_load_ptr(&pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.addedVgInfo);

×

2942

  if (tmpArray && tmpArray == atomic_val_compare_exchange_ptr(&pTaskInfo->pStreamRuntimeInfo->vtableDeployInfo.addedVgInfo, tmpArray, NULL)) {

×

2943

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

×

2944

      SStreamTaskAddr* pTaskAddr = (SStreamTaskAddr*)taosArrayGet(tmpArray, i);

×

2945

      code = taosHashPut(pVtbScan->existOrgTbVg, &pTaskAddr->nodeId, sizeof(pTaskAddr->nodeId), NULL, 0);

×

2946

      QUERY_CHECK_CODE(code, line, _return);

×

2947

      if (pVtbScan->newAddedVgInfo == NULL) {

×

2948

        pVtbScan->newAddedVgInfo = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

2949

        QUERY_CHECK_NULL(pVtbScan->newAddedVgInfo, code, line, _return, terrno)

×

2951

      code = taosHashPut(pVtbScan->newAddedVgInfo, &pTaskAddr->nodeId, sizeof(pTaskAddr->nodeId), pTaskAddr, sizeof(SStreamTaskAddr));

×

2952

      QUERY_CHECK_CODE(code, line, _return);

×

2954

    pVtbScan->needRedeploy = false;

×

2956

    code = TSDB_CODE_STREAM_VTABLE_NEED_REDEPLOY;

×

2957

    QUERY_CHECK_CODE(code, line, _return);

×

2960

_return:

×

2961

  taosArrayClear(tmpArray);

×

2962

  taosArrayDestroy(tmpArray);

×

2963

  if (code) {

×

2964

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

×

2966

  return code;

×

2969

int32_t virtualTableScanBuildDownStreamOpParam(SOperatorInfo* pOperator, tb_uid_t uid, int32_t vgId) {

×

2970

  int32_t                    code = TSDB_CODE_SUCCESS;

×

2971

  int32_t                    line = 0;

×

2972

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

2973

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

2974

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

2976

  pVtbScan->vtbScanParam = NULL;

×

2977

  code = buildVtbScanOperatorParam(pInfo, &pVtbScan->vtbScanParam, uid);

×

2978

  QUERY_CHECK_CODE(code, line, _return);

×

2980

  void* pIter = taosHashIterate(pVtbScan->otbNameToOtbInfoMap, NULL);

×

2981

  while (pIter != NULL) {

×

2982

    SOrgTbInfo*      pMap = (SOrgTbInfo*)pIter;

×

2983

    SOperatorParam*  pExchangeParam = NULL;

×

2984

    SStreamTaskAddr* addr = taosHashGet(pVtbScan->newAddedVgInfo, &pMap->vgId, sizeof(pMap->vgId));

×

2985

    if (addr != NULL) {

×

2986

      SDownstreamSourceNode newSource = {0};

×

2987

      newSource.type = QUERY_NODE_DOWNSTREAM_SOURCE;

×

2988

      newSource.clientId = pTaskInfo->id.taskId;// current task's taskid

×

2989

      newSource.taskId = addr->taskId;

×

2990

      newSource.fetchMsgType = TDMT_STREAM_FETCH;

×

2991

      newSource.localExec = false;

×

2992

      newSource.addr.nodeId = addr->nodeId;

×

2993

      memcpy(&newSource.addr.epSet, &addr->epset, sizeof(SEpSet));

×

2995

      code = buildExchangeOperatorParamForVScan(&pExchangeParam, 0, pMap, &newSource);

×

2996

      QUERY_CHECK_CODE(code, line, _return);

×

2997

      code = taosHashRemove(pVtbScan->newAddedVgInfo, &pMap->vgId, sizeof(pMap->vgId));

×

2998

      QUERY_CHECK_CODE(code, line, _return);

×

3000

      code = buildExchangeOperatorParamForVScan(&pExchangeParam, 0, pMap, NULL);

×

3001

      QUERY_CHECK_CODE(code, line, _return);

×

3003

    QUERY_CHECK_NULL(taosArrayPush(((SVTableScanOperatorParam*)pVtbScan->vtbScanParam->value)->pOpParamArray, &pExchangeParam), code, line, _return, terrno)

×

3004

    pIter = taosHashIterate(pVtbScan->otbNameToOtbInfoMap, pIter);

×

3007

  SOperatorParam*  pExchangeParam = NULL;

×

3008

  code = buildExchangeOperatorParamForVTagScan(&pExchangeParam, 0, vgId, uid);

×

3009

  QUERY_CHECK_CODE(code, line, _return);

×

3010

  ((SVTableScanOperatorParam*)pVtbScan->vtbScanParam->value)->pTagScanOp = pExchangeParam;

×

3012

_return:

×

3013

  if (code) {

×

3014

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

×

3016

  return code;

×

3019

int32_t virtualTableScanGetNext(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

3020

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3021

  int32_t                    line = 0;

×

3022

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

3023

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

3024

  SOperatorInfo*             pVtbScanOp = pOperator->pDownstream[0];

×

3026

  pVtbScan->otbNameToOtbInfoMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

3027

  QUERY_CHECK_NULL(pVtbScan->otbNameToOtbInfoMap, code, line, _return, terrno)

×

3028

  taosHashSetFreeFp(pVtbScan->otbNameToOtbInfoMap, destroySOrgTbInfo);

×

3031

    if (pVtbScan->curTableIdx == pVtbScan->lastTableIdx) {

×

3032

      code = pVtbScanOp->fpSet.getNextFn(pVtbScanOp, pRes);

×

3033

      QUERY_CHECK_CODE(code, line, _return);

×

3035

      taosHashClear(pVtbScan->otbNameToOtbInfoMap);

×

3036

      SArray* pColRefInfo = NULL;

×

3037

      if (pVtbScan->isSuperTable) {

×

3038

        pColRefInfo = (SArray*)taosArrayGetP(pVtbScan->childTableList, pVtbScan->curTableIdx);

×

3040

        pColRefInfo = pInfo->vtbScan.colRefInfo;

×

3042

      QUERY_CHECK_NULL(pColRefInfo, code, line, _return, terrno)

×

3044

      tb_uid_t  uid = 0;

×

3045

      int32_t   vgId = 0;

×

3046

      SHashObj* refMap = NULL;

×

3047

      code = virtualTableScanProcessColRefInfo(pOperator, pColRefInfo, &uid, &vgId, &refMap);

×

3048

      QUERY_CHECK_CODE(code, line, _return);

×

3050

      qDebug("virtual table scan process subtable idx:%d uid:%" PRIu64 " vgId:%d", pVtbScan->curTableIdx, uid, vgId);

×

3052

      code = buildRefSlotGroupsFromRefMap(refMap, pVtbScan->readColList, &pVtbScan->refColGroups);

×

3053

      QUERY_CHECK_CODE(code, line, _return);

×

3055

      code = virtualTableScanBuildDownStreamOpParam(pOperator, uid, vgId);

×

3056

      QUERY_CHECK_CODE(code, line, _return);

×

3059

      pVtbScanOp->status = OP_NOT_OPENED;

×

3060

      code = pVtbScanOp->fpSet.getNextExtFn(pVtbScanOp, pVtbScan->vtbScanParam, pRes);

×

3061

      QUERY_CHECK_CODE(code, line, _return);

×

3064

    if (*pRes) {

×

3066

      pVtbScan->lastTableIdx = pVtbScan->curTableIdx;

×

3067

      break;

×

3070

      pVtbScan->curTableIdx++;

×

3071

      if (pVtbScan->isSuperTable) {

×

3072

        if (pVtbScan->curTableIdx >= taosArrayGetSize(pVtbScan->childTableList)) {

×

3073

          setOperatorCompleted(pOperator);

×

3074

          break;

×

3077

        setOperatorCompleted(pOperator);

×

3078

        break;

×

3083

_return:

×

3084

  taosHashCleanup(pVtbScan->otbNameToOtbInfoMap);

×

3085

  pVtbScan->otbNameToOtbInfoMap = NULL;

×

3086

  if (code) {

×

3087

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

×

3089

  return code;

×

3092

int32_t vtbScanOpen(SOperatorInfo* pOperator) {

×

3093

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3094

  int32_t                    line = 0;

×

3095

  int64_t                    st = 0;

×

3096

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

3097

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

3099

  if (OPTR_IS_OPENED(pOperator)) {

×

3100

    return code;

×

3103

  if (pOperator->cost.openCost == 0) {

×

3104

    st = taosGetTimestampUs();

×

3107

  if (pVtbScan->isSuperTable) {

×

3108

    code = buildVirtualSuperTableScanChildTableMap(pOperator);

×

3109

    QUERY_CHECK_CODE(code, line, _return);

×

3111

    code = buildVirtualNormalChildTableScanChildTableMap(pOperator);

×

3112

    QUERY_CHECK_CODE(code, line, _return);

×

3115

  OPTR_SET_OPENED(pOperator);

×

3117

_return:

×

3118

  if (pOperator->cost.openCost == 0) {

×

3119

    pOperator->cost.openCost = (double)(taosGetTimestampUs() - st) / 1000.0;

×

3121

  if (code) {

×

3122

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

×

3123

    pOperator->pTaskInfo->code = code;

×

3124

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

3126

  return code;

×

3129

int32_t vtbScanNext(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

3130

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3131

  int32_t                    line = 0;

×

3132

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

3133

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

3135

  QRY_PARAM_CHECK(pRes);

×

3136

  if (pOperator->status == OP_EXEC_DONE && !pOperator->pOperatorGetParam) {

×

3137

    return code;

×

3139

  if (pOperator->pOperatorGetParam) {

×

3140

    if (pOperator->status == OP_EXEC_DONE) {

×

3141

      pOperator->status = OP_OPENED;

×

3143

    pVtbScan->curTableIdx = 0;

×

3144

    pVtbScan->lastTableIdx = -1;

×

3145

    pVtbScan->window = ((SDynQueryCtrlOperatorParam *)(pOperator->pOperatorGetParam)->value)->window;

×

3146

    pOperator->pOperatorGetParam = NULL;

×

3148

    pVtbScan->window.skey = INT64_MAX;

×

3149

    pVtbScan->window.ekey = INT64_MIN;

×

3152

  if (pVtbScan->needRedeploy) {

×

3153

    code = virtualTableScanCheckNeedRedeploy(pOperator);

×

3154

    QUERY_CHECK_CODE(code, line, _return);

×

3157

  code = pOperator->fpSet._openFn(pOperator);

×

3158

  QUERY_CHECK_CODE(code, line, _return);

×

3160

  if (pVtbScan->isSuperTable && taosArrayGetSize(pVtbScan->childTableList) == 0) {

×

3161

    setOperatorCompleted(pOperator);

×

3162

    return code;

×

3165

  code = virtualTableScanGetNext(pOperator, pRes);

×

3166

  QUERY_CHECK_CODE(code, line, _return);

×

3168

  return code;

×

3170

_return:

×

3171

  if (code) {

×

3172

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

×

3173

    pOperator->pTaskInfo->code = code;

×

3174

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

3176

  return code;

×

3186

static int32_t vtbDefaultOpen(SOperatorInfo* pOperator) {

×

3187

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3188

  int32_t                    line = 0;

×

3189

  int64_t                    st = 0;

×

3190

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

3191

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

3193

  if (OPTR_IS_OPENED(pOperator)) {

×

3194

    return code;

×

3197

  if (pOperator->cost.openCost == 0) {

×

3198

    st = taosGetTimestampUs();

×

3201

  code = buildVirtualSuperTableScanChildTableMap(pOperator);

×

3202

  QUERY_CHECK_CODE(code, line, _return);

×

3204

  OPTR_SET_OPENED(pOperator);

×

3206

_return:

×

3207

  if (pOperator->cost.openCost == 0) {

×

3208

    pOperator->cost.openCost = (double)(taosGetTimestampUs() - st) / 1000.0;

×

3210

  if (code) {

×

3211

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

×

3212

    pOperator->pTaskInfo->code = code;

×

3213

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

3215

  return code;

×

3226

int32_t vtbTsScanNext(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

3227

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3228

  int32_t                    line = 0;

×

3229

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

3230

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

3231

  SOperatorInfo*             pMergeOp = pOperator->pDownstream[0];

×

3232

  SOperatorParam*            pMergeParam = NULL;

×

3233

  SSDataBlock*               pResult = NULL;

×

3235

  QRY_PARAM_CHECK(pRes);

×

3236

  if (pOperator->status == OP_EXEC_DONE) {

×

3237

    return code;

×

3240

  code = pOperator->fpSet._openFn(pOperator);

×

3241

  QUERY_CHECK_CODE(code, line, _return);

×

3243

  if (pVtbScan->genNewParam) {

×

3244

    code = buildMergeOperatorParamForTsScan(pInfo, &pMergeParam);

×

3245

    QUERY_CHECK_CODE(code, line, _return);

×

3247

    code = pMergeOp->fpSet.getNextExtFn(pMergeOp, pMergeParam, &pResult);

×

3248

    QUERY_CHECK_CODE(code, line, _return);

×

3249

    pVtbScan->genNewParam = false;

×

3251

    code = pMergeOp->fpSet.getNextFn(pMergeOp, &pResult);

×

3252

    QUERY_CHECK_CODE(code, line, _return);

×

3255

  if (pResult) {

×

3256

    *pRes = pResult;

×

3258

    *pRes = NULL;

×

3259

    setOperatorCompleted(pOperator);

×

3262

_return:

×

3263

  if (code) {

×

3264

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

×

3265

    pOperator->pTaskInfo->code = code;

×

3266

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

3268

  return code;

×

3271

int32_t initSeqStbJoinTableHash(SStbJoinPrevJoinCtx* pPrev, bool batchFetch) {

×

3272

  if (batchFetch) {

×

3273

    pPrev->leftHash = tSimpleHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));

×

3274

    if (NULL == pPrev->leftHash) {

×

3275

      return terrno;

×

3277

    pPrev->rightHash = tSimpleHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));

×

3278

    if (NULL == pPrev->rightHash) {

×

3279

      return terrno;

×

3282

    pPrev->leftCache = tSimpleHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

×

3283

    if (NULL == pPrev->leftCache) {

×

3284

      return terrno;

×

3286

    pPrev->rightCache = tSimpleHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

×

3287

    if (NULL == pPrev->rightCache) {

×

3288

      return terrno;

×

3290

    pPrev->onceTable = tSimpleHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));

×

3291

    if (NULL == pPrev->onceTable) {

×

3292

      return terrno;

×

3296

  return TSDB_CODE_SUCCESS;

×

3299

static void updateDynTbUidIfNeeded(SVtbScanDynCtrlInfo* pVtbScan, SStreamRuntimeInfo* pStreamRuntimeInfo) {

×

3300

  if (pStreamRuntimeInfo == NULL) {

×

3301

    return;

×

3304

  SArray* vals = pStreamRuntimeInfo->funcInfo.pStreamPartColVals;

×

3305

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

×

3306

    SStreamGroupValue* pValue = taosArrayGet(vals, i);

×

3307

    if (pValue != NULL && pValue->isTbname && pValue->uid != pVtbScan->dynTbUid) {

×

3308

      qTrace("dynQueryCtrl dyn tb uid:%" PRIu64 " reset to:%" PRIu64, pVtbScan->dynTbUid, pValue->uid);

×

3310

      pVtbScan->dynTbUid = pValue->uid;

×

3311

      break;

×

3316

static int32_t initVtbScanInfo(SDynQueryCtrlOperatorInfo* pInfo, SMsgCb* pMsgCb,

×

3318

  int32_t      code = TSDB_CODE_SUCCESS;

×

3319

  int32_t      line = 0;

×

3321

  code = tsem_init(&pInfo->vtbScan.ready, 0, 0);

×

3322

  QUERY_CHECK_CODE(code, line, _return);

×

3324

  pInfo->vtbScan.genNewParam = true;

×

3325

  pInfo->vtbScan.batchProcessChild = pPhyciNode->vtbScan.batchProcessChild;

×

3326

  pInfo->vtbScan.hasPartition = pPhyciNode->vtbScan.hasPartition;

×

3327

  pInfo->vtbScan.scanAllCols = pPhyciNode->vtbScan.scanAllCols;

×

3328

  pInfo->vtbScan.isSuperTable = pPhyciNode->vtbScan.isSuperTable;

×

3329

  pInfo->vtbScan.rversion = pPhyciNode->vtbScan.rversion;

×

3330

  pInfo->vtbScan.uid = pPhyciNode->vtbScan.uid;

×

3331

  pInfo->vtbScan.suid = pPhyciNode->vtbScan.suid;

×

3332

  pInfo->vtbScan.epSet = pPhyciNode->vtbScan.mgmtEpSet;

×

3333

  pInfo->vtbScan.acctId = pPhyciNode->vtbScan.accountId;

×

3334

  pInfo->vtbScan.needRedeploy = false;

×

3335

  pInfo->vtbScan.pMsgCb = pMsgCb;

×

3336

  pInfo->vtbScan.curTableIdx = 0;

×

3337

  pInfo->vtbScan.lastTableIdx = -1;

×

3338

  pInfo->vtbScan.dynTbUid = 0;

×

3339

  pInfo->vtbScan.dbName = taosStrdup(pPhyciNode->vtbScan.dbName);

×

3340

  pInfo->vtbScan.tbName = taosStrdup(pPhyciNode->vtbScan.tbName);

×

3341

  QUERY_CHECK_NULL(pInfo->vtbScan.dbName, code, line, _return, terrno)

×

3342

  QUERY_CHECK_NULL(pInfo->vtbScan.tbName, code, line, _return, terrno)

×

3343

  pInfo->vtbScan.existOrgTbVg = taosHashInit(1, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);

×

3344

  QUERY_CHECK_NULL(pInfo->vtbScan.existOrgTbVg, code, line, _return, terrno)

×

3345

  SNode* node = NULL;

×

3346

  FOREACH(node, pPhyciNode->vtbScan.pOrgVgIds) {

×

3347

    SValueNode* valueNode = (SValueNode*)node;

×

3348

    int32_t vgId = (int32_t)valueNode->datum.i;

×

3349

    code = taosHashPut(pInfo->vtbScan.existOrgTbVg, &vgId, sizeof(vgId), NULL, 0);

×

3350

    QUERY_CHECK_CODE(code, line, _return);

×

3353

  if (pPhyciNode->dynTbname && pTaskInfo) {

×

3354

    updateDynTbUidIfNeeded(&pInfo->vtbScan, pTaskInfo->pStreamRuntimeInfo);

×

3357

  pInfo->vtbScan.readColList = taosArrayInit(LIST_LENGTH(pPhyciNode->vtbScan.pScanCols), sizeof(col_id_t));

×

3358

  QUERY_CHECK_NULL(pInfo->vtbScan.readColList, code, line, _return, terrno)

×

3360

  SNode* colNode = NULL;

×

3361

  FOREACH(colNode, pPhyciNode->vtbScan.pScanCols) {

×

3362

    SColumnNode* pNode = (SColumnNode*)colNode;

×

3363

    QUERY_CHECK_NULL(pNode, code, line, _return, terrno)

×

3364

    QUERY_CHECK_NULL(taosArrayPush(pInfo->vtbScan.readColList, &pNode->colId), code, line, _return, terrno)

×

3367

  pInfo->vtbScan.readColSet =

×

3368

      taosHashInit(taosArrayGetSize(pInfo->vtbScan.readColList) > 0 ? taosArrayGetSize(pInfo->vtbScan.readColList) : 1,

×

3370

  QUERY_CHECK_NULL(pInfo->vtbScan.readColSet, code, line, _return, terrno)

×

3371

  for (int32_t i = 0; i < taosArrayGetSize(pInfo->vtbScan.readColList); i++) {

×

3372

    col_id_t colId = *(col_id_t*)taosArrayGet(pInfo->vtbScan.readColList, i);

×

3373

    code = taosHashPut(pInfo->vtbScan.readColSet, &colId, sizeof(colId), NULL, 0);

×

3374

    QUERY_CHECK_CODE(code, line, _return);

×

3377

  pInfo->vtbScan.refColGroups = NULL;

×

3379

  pInfo->vtbScan.childTableList = taosArrayInit(10, POINTER_BYTES);

×

3380

  QUERY_CHECK_NULL(pInfo->vtbScan.childTableList, code, line, _return, terrno)

×

3382

  pInfo->vtbScan.dbVgInfoMap = taosHashInit(taosArrayGetSize(pInfo->vtbScan.childTableList), taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);

×

3383

  QUERY_CHECK_NULL(pInfo->vtbScan.dbVgInfoMap, code, line, _return, terrno)

×

3385

  pInfo->vtbScan.otbNameToOtbInfoMap = NULL;

×

3386

  pInfo->vtbScan.otbVgIdToOtbInfoArrayMap = NULL;

×

3387

  pInfo->vtbScan.vtbUidToVgIdMapMap = NULL;

×

3388

  pInfo->vtbScan.vtbGroupIdToVgIdMapMap = NULL;

×

3389

  pInfo->vtbScan.vtbUidTagListMap = NULL;

×

3390

  pInfo->vtbScan.vtbGroupIdTagListMap = NULL;

×

3391

  pInfo->vtbScan.vtbUidToGroupIdMap = NULL;

×

3393

  return code;

×

3394

_return:

×

3397

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

×

3398

  return code;

×

3401

static int32_t initVtbWindowInfo(SDynQueryCtrlOperatorInfo* pInfo, SDynQueryCtrlPhysiNode* pPhyciNode,

×

3403

  int32_t              code = TSDB_CODE_SUCCESS;

×

3404

  int32_t              line = 0;

×

3405

  SDataBlockDescNode*  pDescNode = pPhyciNode->node.pOutputDataBlockDesc;

×

3407

  pInfo->vtbWindow.wstartSlotId = pPhyciNode->vtbWindow.wstartSlotId;

×

3408

  pInfo->vtbWindow.wendSlotId = pPhyciNode->vtbWindow.wendSlotId;

×

3409

  pInfo->vtbWindow.wdurationSlotId = pPhyciNode->vtbWindow.wdurationSlotId;

×

3410

  pInfo->vtbWindow.pTargets = pPhyciNode->vtbWindow.pTargets;

×

3411

  pInfo->vtbWindow.isVstb = pPhyciNode->vtbWindow.isVstb;

×

3412

  pInfo->vtbWindow.extendOption = pPhyciNode->vtbWindow.extendOption;

×

3414

  pInfo->vtbWindow.pRes = createDataBlockFromDescNode(pDescNode);

×

3415

  QUERY_CHECK_NULL(pInfo->vtbWindow.pRes, code, line, _return, terrno)

×

3417

  pInfo->vtbWindow.pWins = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);

×

3418

  QUERY_CHECK_NULL(pInfo->vtbWindow.pWins, code, line, _return, terrno)

×

3420

  pInfo->vtbWindow.outputWstartSlotId = -1;

×

3421

  pInfo->vtbWindow.outputWendSlotId = -1;

×

3422

  pInfo->vtbWindow.outputWdurationSlotId = -1;

×

3423

  pInfo->vtbWindow.curWinBatchIdx = 0;

×

3425

  initResultSizeInfo(&pOperator->resultInfo, 1);

×

3426

  code = blockDataEnsureCapacity(pInfo->vtbWindow.pRes, pOperator->resultInfo.capacity);

×

3427

  QUERY_CHECK_CODE(code, line, _return);

×

3429

  return code;

×

3430

_return:

×

3431

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

×

3432

  return code;

×

3435

static int32_t extractTsCol(SSDataBlock* pBlock, int32_t slotId, TSKEY** ppTsCols) {

×

3436

  int32_t code = TSDB_CODE_SUCCESS;

×

3437

  int32_t lino = 0;

×

3439

  if (pBlock->pDataBlock != NULL && pBlock->info.dataLoad) {

×

3440

    SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, slotId);

×

3441

    QUERY_CHECK_NULL(pColDataInfo, code, lino, _return, terrno)

×

3443

    *ppTsCols = (int64_t*)pColDataInfo->pData;

×

3445

    if ((*ppTsCols)[0] != 0 && (pBlock->info.window.skey == 0 && pBlock->info.window.ekey == 0)) {

×

3446

      code = blockDataUpdateTsWindow(pBlock, slotId);

×

3447

      QUERY_CHECK_CODE(code, lino, _return);

×

3451

  return code;

×

3452

_return:

×

3453

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

×

3454

  return code;

×

3457

int32_t vtbWindowOpen(SOperatorInfo* pOperator) {

×

3458

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3459

  int32_t                    lino = 0;

×

3460

  SDynQueryCtrlOperatorInfo* pDynInfo = pOperator->info;

×

3461

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

3462

  SVtbWindowDynCtrlInfo*     pInfo = &pDynInfo->vtbWindow;

×

3463

  int64_t                    st = 0;

×

3465

  if (OPTR_IS_OPENED(pOperator)) {

×

3466

    return code;

×

3469

  if (pOperator->cost.openCost == 0) {

×

3470

    st = taosGetTimestampUs();

×

3473

  while (1) {

×

3474

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, 0);

×

3475

    if (pBlock == NULL) {

×

3476

      break;

×

3479

    if (pInfo->outputWendSlotId == -1 && pInfo->outputWstartSlotId == -1 && pInfo->outputWdurationSlotId == -1) {

×

3480

      for (int32_t i = 0; i < LIST_LENGTH(pInfo->pTargets); ++i) {

×

3481

        STargetNode* pNode = (STargetNode*)nodesListGetNode(pInfo->pTargets, i);

×

3482

        if (nodeType(pNode->pExpr) == QUERY_NODE_COLUMN && ((SColumnNode*)pNode->pExpr)->dataBlockId == pBlock->info.id.blockId) {

×

3483

          if (((SColumnNode*)pNode->pExpr)->slotId == pDynInfo->vtbWindow.wstartSlotId) {

×

3484

            pInfo->outputWstartSlotId = i;

×

3485

          } else if (((SColumnNode*)pNode->pExpr)->slotId == pDynInfo->vtbWindow.wendSlotId) {

×

3486

            pInfo->outputWendSlotId = i;

×

3487

          } else if (((SColumnNode*)pNode->pExpr)->slotId == pDynInfo->vtbWindow.wdurationSlotId) {

×

3488

            pInfo->outputWdurationSlotId = i;

×

3494

    TSKEY* wstartCol = NULL;

×

3495

    TSKEY* wendCol = NULL;

×

3497

    code = extractTsCol(pBlock, pDynInfo->vtbWindow.wstartSlotId, &wstartCol);

×

3498

    QUERY_CHECK_CODE(code, lino, _return);

×

3499

    code = extractTsCol(pBlock, pDynInfo->vtbWindow.wendSlotId, &wendCol);

×

3500

    QUERY_CHECK_CODE(code, lino, _return);

×

3502

    SArray* pWin = taosArrayInit(pBlock->info.rows, sizeof(SExtWinTimeWindow));

×

3503

    QUERY_CHECK_NULL(pWin, code, lino, _return, terrno)

×

3505

    QUERY_CHECK_NULL(taosArrayReserve(pWin, pBlock->info.rows), code, lino, _return, terrno);

×

3507

    for (int32_t i = 0; i < pBlock->info.rows; i++) {

×

3508

      SExtWinTimeWindow* pWindow = taosArrayGet(pWin, i);

×

3509

      QUERY_CHECK_NULL(pWindow, code, lino, _return, terrno)

×

3510

      pWindow->tw.skey = wstartCol[i];

×

3511

      pWindow->tw.ekey = wendCol[i] + 1;

×

3512

      pWindow->winOutIdx = -1;

×

3515

    QUERY_CHECK_NULL(taosArrayPush(pDynInfo->vtbWindow.pWins, &pWin), code, lino, _return, terrno);

×

3519

  int32_t winBatchNum = (int32_t)taosArrayGetSize(pDynInfo->vtbWindow.pWins);

×

3520

  if (winBatchNum > 0) {

×

3521

    SArray* firstBatch = (SArray*)taosArrayGetP(pDynInfo->vtbWindow.pWins, 0);

×

3522

    SArray* lastBatch = (SArray*)taosArrayGetP(pDynInfo->vtbWindow.pWins, winBatchNum - 1);

×

3524

    QUERY_CHECK_NULL(firstBatch, code, lino, _return, terrno)

×

3525

    QUERY_CHECK_NULL(lastBatch, code, lino, _return, terrno)

×

3527

    SExtWinTimeWindow* firstWin = (SExtWinTimeWindow*)taosArrayGet(firstBatch, 0);

×

3528

    SExtWinTimeWindow* lastWin = (SExtWinTimeWindow*)taosArrayGetLast(lastBatch);

×

3530

    QUERY_CHECK_NULL(firstWin, code, lino, _return, terrno)

×

3531

    QUERY_CHECK_NULL(lastWin, code, lino, _return, terrno)

×

3533

    if (pInfo->extendOption == STATE_WIN_EXTEND_OPTION_BACKWARD) {

×

3534

      lastWin->tw.ekey = INT64_MAX;

×

3536

    if (pInfo->extendOption == STATE_WIN_EXTEND_OPTION_FORWARD) {

×

3537

      firstWin->tw.skey = INT64_MIN;

×

3541

  if (pInfo->isVstb) {

×

3542

    code = buildVirtualSuperTableScanChildTableMap(pOperator);

×

3543

    QUERY_CHECK_CODE(code, lino, _return);

×

3546

  OPTR_SET_OPENED(pOperator);

×

3548

  if (pOperator->cost.openCost == 0) {

×

3549

    pOperator->cost.openCost = (double)(taosGetTimestampUs() - st) / 1000.0;

×

3552

_return:

×

3553

  if (code != TSDB_CODE_SUCCESS) {

×

3554

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

×

3555

    pTaskInfo->code = code;

×

3556

    T_LONG_JMP(pTaskInfo->env, code);

×

3558

  return code;

×

3561

static int32_t buildExternalWindowOperatorParamEx(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes, SArray* pWins, int32_t idx) {

×

3562

  int32_t                       code = TSDB_CODE_SUCCESS;

×

3563

  int32_t                       lino = 0;

×

3564

  SExternalWindowOperatorParam* pExtWinOp = NULL;

×

3566

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

3567

  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno)

×

3569

  pExtWinOp = taosMemoryMalloc(sizeof(SExternalWindowOperatorParam));

×

3570

  QUERY_CHECK_NULL(pExtWinOp, code, lino, _return, terrno)

×

3572

  pExtWinOp->ExtWins = taosArrayDup(pWins, NULL);

×

3573

  QUERY_CHECK_NULL(pExtWinOp->ExtWins, code, lino, _return, terrno)

×

3575

  SExtWinTimeWindow* firstWin = (SExtWinTimeWindow*)taosArrayGet(pWins, 0);

×

3576

  SExtWinTimeWindow* lastWin = (SExtWinTimeWindow*)taosArrayGetLast(pWins);

×

3578

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

×

3579

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

×

3581

  (*ppRes)->pChildren = taosArrayInit(1, POINTER_BYTES);

×

3582

  QUERY_CHECK_NULL((*ppRes)->pChildren, code, lino, _return, terrno)

×

3584

  SOperatorParam* pExchangeParam = NULL;

×

3585

  code = buildBatchExchangeOperatorParamForVirtual(

×

3587

      (STimeWindow){.skey = firstWin->tw.skey, .ekey = lastWin->tw.ekey}, EX_SRC_TYPE_VSTB_WIN_SCAN,

×

3589

  QUERY_CHECK_CODE(code, lino, _return);

×

3591

  QUERY_CHECK_NULL(taosArrayPush((*ppRes)->pChildren, &pExchangeParam), code, lino, _return, terrno)

×

3593

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_EXTERNAL_WINDOW;

×

3594

  (*ppRes)->downstreamIdx = idx;

×

3595

  (*ppRes)->value = pExtWinOp;

×

3596

  (*ppRes)->reUse = false;

×

3598

  return code;

×

3599

_return:

×

3600

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

×

3601

  if (pExtWinOp) {

×

3602

    if (pExtWinOp->ExtWins) {

×

3603

      taosArrayDestroy(pExtWinOp->ExtWins);

×

3605

    taosMemoryFree(pExtWinOp);

×

3607

  if (*ppRes) {

×

3608

    if ((*ppRes)->pChildren) {

×

3609

      for (int32_t i = 0; i < taosArrayGetSize((*ppRes)->pChildren); i++) {

×

3610

        SOperatorParam* pChildParam = taosArrayGetP((*ppRes)->pChildren, i);

×

3611

        if (pChildParam) {

×

3612

          freeOperatorParam(pChildParam, OP_GET_PARAM);

×

3615

      taosArrayDestroy((*ppRes)->pChildren);

×

3617

    taosMemoryFree(*ppRes);

×

3618

    *ppRes = NULL;

×

3620

  return code;

×

3623

int32_t vtbWindowNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {

×

3624

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3625

  int32_t                    lino = 0;

×

3626

  SDynQueryCtrlOperatorInfo* pDynInfo = pOperator->info;

×

3627

  SExecTaskInfo*             pTaskInfo = pOperator->pTaskInfo;

×

3628

  int64_t                    st = taosGetTimestampUs();

×

3629

  int32_t                    numOfWins = 0;

×

3630

  SOperatorInfo*             mergeOp = NULL;

×

3631

  SOperatorInfo*             extWinOp = NULL;

×

3632

  SOperatorParam*            pMergeParam = NULL;

×

3633

  SOperatorParam*            pExtWinParam = NULL;

×

3634

  SVtbWindowDynCtrlInfo*     pInfo = &pDynInfo->vtbWindow;

×

3635

  SSDataBlock*               pRes = pInfo->pRes;

×

3637

  code = pOperator->fpSet._openFn(pOperator);

×

3638

  QUERY_CHECK_CODE(code, lino, _return);

×

3640

  if (pInfo->curWinBatchIdx >= taosArrayGetSize(pInfo->pWins)) {

×

3641

    *ppRes = NULL;

×

3642

    return code;

×

3645

  SArray* pWinArray = (SArray*)taosArrayGetP(pInfo->pWins, pInfo->curWinBatchIdx);

×

3646

  QUERY_CHECK_NULL(pWinArray, code, lino, _return, terrno)

×

3648

  numOfWins = (int32_t)taosArrayGetSize(pWinArray);

×

3650

  if (pInfo->isVstb) {

×

3651

    extWinOp = pOperator->pDownstream[2];

×

3652

    code = buildExternalWindowOperatorParamEx(pDynInfo, &pExtWinParam, pWinArray, extWinOp->numOfDownstream);

×

3653

    QUERY_CHECK_CODE(code, lino, _return);

×

3655

    SSDataBlock* pExtWinBlock = NULL;

×

3656

    code = extWinOp->fpSet.getNextExtFn(extWinOp, pExtWinParam, &pExtWinBlock);

×

3657

    QUERY_CHECK_CODE(code, lino, _return);

×

3658

    setOperatorCompleted(extWinOp);

×

3660

    if (extWinOp->pOperatorGetParam) {

×

3661

      freeOperatorParam(extWinOp->pOperatorGetParam, OP_GET_PARAM);

×

3662

      extWinOp->pOperatorGetParam = NULL;

×

3665

    if (extWinOp->pDownstreamGetParams) {

×

3666

      for (int32_t i = 0; i < extWinOp->numOfDownstream; i++) {

×

3667

        if (extWinOp->pDownstreamGetParams[i]) {

×

3668

          freeOperatorParam(extWinOp->pDownstreamGetParams[i], OP_GET_PARAM);

×

3669

          extWinOp->pDownstreamGetParams[i] = NULL;

×

3674

    blockDataCleanup(pRes);

×

3675

    code = blockDataEnsureCapacity(pRes, numOfWins);

×

3676

    QUERY_CHECK_CODE(code, lino, _return);

×

3678

    if (pExtWinBlock) {

×

3679

      code = copyColumnsValue(pInfo->pTargets, pExtWinBlock->info.id.blockId, pRes, pExtWinBlock, numOfWins);

×

3680

      QUERY_CHECK_CODE(code, lino, _return);

×

3682

      if (pInfo->curWinBatchIdx == 0) {

×

3684

        SExtWinTimeWindow* firstWin = (SExtWinTimeWindow*)taosArrayGet(taosArrayGetP(pInfo->pWins, 0), 0);

×

3685

        QUERY_CHECK_NULL(firstWin, code, lino, _return, terrno)

×

3687

        firstWin->tw.skey = TMAX(firstWin->tw.skey, pExtWinBlock->info.window.skey);

×

3689

      if (pInfo->curWinBatchIdx == taosArrayGetSize(pInfo->pWins) - 1) {

×

3691

        SExtWinTimeWindow* lastWin = (SExtWinTimeWindow*)taosArrayGetLast(taosArrayGetP(pInfo->pWins, taosArrayGetSize(pInfo->pWins) - 1));

×

3692

        QUERY_CHECK_NULL(lastWin, code, lino, _return, terrno)

×

3694

        lastWin->tw.ekey = TMIN(lastWin->tw.ekey, pExtWinBlock->info.window.ekey + 1);

×

3698

    mergeOp = pOperator->pDownstream[1];

×

3699

    code = buildMergeOperatorParam(pDynInfo, &pMergeParam, pWinArray, mergeOp->numOfDownstream, numOfWins);

×

3700

    QUERY_CHECK_CODE(code, lino, _return);

×

3702

    SSDataBlock* pMergedBlock = NULL;

×

3703

    code = mergeOp->fpSet.getNextExtFn(mergeOp, pMergeParam, &pMergedBlock);

×

3704

    QUERY_CHECK_CODE(code, lino, _return);

×

3706

    if (mergeOp->pOperatorGetParam) {

×

3707

      freeOperatorParam(mergeOp->pOperatorGetParam, OP_GET_PARAM);

×

3708

      mergeOp->pOperatorGetParam = NULL;

×

3711

    if (mergeOp->pDownstreamGetParams) {

×

3712

      for (int32_t i = 0; i < mergeOp->numOfDownstream; i++) {

×

3713

        if (mergeOp->pDownstreamGetParams[i]) {

×

3714

          freeOperatorParam(mergeOp->pDownstreamGetParams[i], OP_GET_PARAM);

×

3715

          mergeOp->pDownstreamGetParams[i] = NULL;

×

3720

    blockDataCleanup(pRes);

×

3721

    code = blockDataEnsureCapacity(pRes, numOfWins);

×

3722

    QUERY_CHECK_CODE(code, lino, _return);

×

3724

    if (pMergedBlock) {

×

3725

      code = copyColumnsValue(pInfo->pTargets, pMergedBlock->info.id.blockId, pRes, pMergedBlock, numOfWins);

×

3726

      QUERY_CHECK_CODE(code, lino, _return);

×

3728

      if (pInfo->curWinBatchIdx == 0) {

×

3730

        SExtWinTimeWindow* firstWin = (SExtWinTimeWindow*)taosArrayGet(taosArrayGetP(pInfo->pWins, 0), 0);

×

3731

        QUERY_CHECK_NULL(firstWin, code, lino, _return, terrno)

×

3733

        firstWin->tw.skey = TMAX(firstWin->tw.skey, pMergedBlock->info.window.skey);

×

3735

      if (pInfo->curWinBatchIdx == taosArrayGetSize(pInfo->pWins) - 1) {

×

3737

        SExtWinTimeWindow* lastWin = (SExtWinTimeWindow*)taosArrayGetLast(taosArrayGetP(pInfo->pWins, taosArrayGetSize(pInfo->pWins) - 1));

×

3738

        QUERY_CHECK_NULL(lastWin, code, lino, _return, terrno)

×

3740

        lastWin->tw.ekey = TMIN(lastWin->tw.ekey, pMergedBlock->info.window.ekey + 1);

×

3746

  if (pInfo->outputWstartSlotId != -1) {

×

3747

    SColumnInfoData* pWstartCol = taosArrayGet(pRes->pDataBlock, pInfo->outputWstartSlotId);

×

3748

    QUERY_CHECK_NULL(pWstartCol, code, lino, _return, terrno)

×

3750

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

×

3751

      SExtWinTimeWindow* pWindow = (SExtWinTimeWindow*)taosArrayGet(pWinArray, i);

×

3752

      QUERY_CHECK_NULL(pWindow, code, lino, _return, terrno)

×

3753

      code = colDataSetVal(pWstartCol, i, (const char*)&pWindow->tw.skey, false);

×

3754

      QUERY_CHECK_CODE(code, lino, _return);

×

3757

  if (pInfo->outputWendSlotId != -1) {

×

3758

    SColumnInfoData* pWendCol = taosArrayGet(pRes->pDataBlock, pInfo->outputWendSlotId);

×

3759

    QUERY_CHECK_NULL(pWendCol, code, lino, _return, terrno)

×

3761

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

×

3762

      SExtWinTimeWindow* pWindow = (SExtWinTimeWindow*)taosArrayGet(pWinArray, i);

×

3763

      QUERY_CHECK_NULL(pWindow, code, lino, _return, terrno)

×

3764

      TSKEY ekey = pWindow->tw.ekey - 1;

×

3765

      code = colDataSetVal(pWendCol, i, (const char*)&ekey, false);

×

3766

      QUERY_CHECK_CODE(code, lino, _return);

×

3769

  if (pInfo->outputWdurationSlotId != -1) {

×

3770

    SColumnInfoData* pWdurationCol = taosArrayGet(pRes->pDataBlock, pInfo->outputWdurationSlotId);

×

3771

    QUERY_CHECK_NULL(pWdurationCol, code, lino, _return, terrno)

×

3773

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

×

3774

      SExtWinTimeWindow* pWindow = (SExtWinTimeWindow*)taosArrayGet(pWinArray, i);

×

3775

      QUERY_CHECK_NULL(pWindow, code, lino, _return, terrno)

×

3776

      int64_t duration = pWindow->tw.ekey - 1 - pWindow->tw.skey;

×

3777

      code = colDataSetVal(pWdurationCol, i, (const char*)&duration, false);

×

3778

      QUERY_CHECK_CODE(code, lino, _return);

×

3782

  pRes->info.rows = numOfWins;

×

3783

  *ppRes = pRes;

×

3784

  pInfo->curWinBatchIdx++;

×

3786

  return code;

×

3788

_return:

×

3789

  if (code != TSDB_CODE_SUCCESS) {

×

3790

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

×

3791

    pTaskInfo->code = code;

×

3792

    T_LONG_JMP(pTaskInfo->env, code);

×

3794

  return code;

×

3797

static int32_t resetDynQueryCtrlOperState(SOperatorInfo* pOper) {

×

3798

  SDynQueryCtrlOperatorInfo*    pDyn = pOper->info;

×

3799

  SDynQueryCtrlPhysiNode const* pPhyciNode = pOper->pPhyNode;

×

3800

  SExecTaskInfo*                pTaskInfo = pOper->pTaskInfo;

×

3802

  pOper->status = OP_NOT_OPENED;

×

3804

  switch (pDyn->qType) {

×

3805

    case DYN_QTYPE_STB_HASH:{

×

3806

      pDyn->stbJoin.execInfo = (SDynQueryCtrlExecInfo){0};

×

3807

      SStbJoinDynCtrlInfo* pStbJoin = &pDyn->stbJoin;

×

3808

      destroyStbJoinDynCtrlInfo(&pDyn->stbJoin);

×

3810

      int32_t code = initSeqStbJoinTableHash(&pDyn->stbJoin.ctx.prev, pDyn->stbJoin.basic.batchFetch);

×

3811

      if (TSDB_CODE_SUCCESS != code) {

×

3812

        qError("initSeqStbJoinTableHash failed since %s", tstrerror(code));

×

3813

        return code;

×

3815

      pStbJoin->ctx.prev.pListHead = NULL;

×

3816

      pStbJoin->ctx.prev.joinBuild = false;

×

3817

      pStbJoin->ctx.prev.pListTail = NULL;

×

3818

      pStbJoin->ctx.prev.tableNum = 0;

×

3820

      pStbJoin->ctx.post = (SStbJoinPostJoinCtx){0};

×

3821

      break;

×

3823

    case DYN_QTYPE_VTB_SCAN:

×

3827

      SVtbScanDynCtrlInfo* pVtbScan = &pDyn->vtbScan;

×

3829

      if (pVtbScan->otbNameToOtbInfoMap) {

×

3830

        taosHashSetFreeFp(pVtbScan->otbNameToOtbInfoMap, destroySOrgTbInfo);

×

3831

        taosHashCleanup(pVtbScan->otbNameToOtbInfoMap);

×

3832

        pVtbScan->otbNameToOtbInfoMap = NULL;

×

3834

      if (pVtbScan->pRsp) {

×

3835

        tFreeSUsedbRsp(pVtbScan->pRsp);

×

3836

        taosMemoryFreeClear(pVtbScan->pRsp);

×

3838

      if (pVtbScan->colRefInfo) {

×

3839

        taosArrayDestroyEx(pVtbScan->colRefInfo, destroyColRefInfo);

×

3840

        pVtbScan->colRefInfo = NULL;

×

3842

      if (pVtbScan->childTableMap) {

×

3843

        taosHashCleanup(pVtbScan->childTableMap);

×

3844

        pVtbScan->childTableMap = NULL;

×

3846

      if (pVtbScan->childTableList) {

×

3847

        taosArrayClearEx(pVtbScan->childTableList, destroyColRefArray);

×

3849

      if (pPhyciNode->dynTbname && pTaskInfo) {

×

3850

        updateDynTbUidIfNeeded(pVtbScan, pTaskInfo->pStreamRuntimeInfo);

×

3852

      pVtbScan->curTableIdx = 0;

×

3853

      pVtbScan->lastTableIdx = -1;

×

3854

      break;

×

3856

    case DYN_QTYPE_VTB_WINDOW: {

×

3857

      SVtbWindowDynCtrlInfo* pVtbWindow = &pDyn->vtbWindow;

×

3858

      if (pVtbWindow->pRes) {

×

3859

        blockDataDestroy(pVtbWindow->pRes);

×

3860

        pVtbWindow->pRes = NULL;

×

3862

      if (pVtbWindow->pWins) {

×

3863

        taosArrayDestroyEx(pVtbWindow->pWins, destroyWinArray);

×

3864

        pVtbWindow->pWins = NULL;

×

3866

      pVtbWindow->outputWdurationSlotId = -1;

×

3867

      pVtbWindow->outputWendSlotId = -1;

×

3868

      pVtbWindow->outputWstartSlotId = -1;

×

3869

      pVtbWindow->curWinBatchIdx = 0;

×

3870

      break;

×

3872

    default:

×

3873

      qError("unsupported dynamic query ctrl type: %d", pDyn->qType);

×

3874

      break;

×

3876

  return 0;

×

3879

int32_t virtualTableAggGetNext(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

3880

  int32_t                    code = TSDB_CODE_SUCCESS;

×

3881

  int32_t                    line = 0;

×

3882

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

3883

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

3884

  SOperatorInfo*             pAggOp = pOperator->pDownstream[pOperator->numOfDownstream - 1];

×

3885

  SOperatorInfo*             pTagScanOp = pOperator->pDownstream[1];

×

3886

  SOperatorParam*            pAggParam = NULL;

×

3888

  if (pInfo->vtbScan.hasPartition) {

×

3889

    if (pInfo->vtbScan.batchProcessChild) {

×

3890

      void* pIter = taosHashIterate(pVtbScan->vtbGroupIdTagListMap, NULL);

×

3891

      while (pIter) {

×

3892

        size_t     keyLen = 0;

×

3893

        uint64_t   groupid = *(uint64_t*)taosHashGetKey(pIter, &keyLen);

×

3895

        code = buildAggOperatorParamWithGroupId(pInfo, groupid, &pAggParam);

×

3896

        QUERY_CHECK_CODE(code, line, _return);

×

3898

        if (pAggParam) {

×

3899

          code = pAggOp->fpSet.getNextExtFn(pAggOp, pAggParam, pRes);

×

3900

          QUERY_CHECK_CODE(code, line, _return);

×

3902

          *pRes = NULL;

×

3905

        pIter = taosHashIterate(pVtbScan->vtbGroupIdTagListMap, pIter);

×

3907

        if (*pRes) {

×

3908

          (*pRes)->info.id.groupId = groupid;

×

3909

          code = taosHashRemove(pVtbScan->vtbGroupIdTagListMap, &groupid, keyLen);

×

3910

          QUERY_CHECK_CODE(code, line, _return);

×

3911

          break;

×

3915

      void *pIter = taosHashIterate(pVtbScan->vtbGroupIdTagListMap, NULL);

×

3916

      while (pIter) {

×

3917

        size_t     keyLen = 0;

×

3918

        uint64_t*  groupid = (uint64_t*)taosHashGetKey(pIter, &keyLen);

×

3919

        SHashObj*  vtbUidTagListMap = *(SHashObj**)pIter;

×

3921

        void* pIter2 = taosHashIterate(vtbUidTagListMap, NULL);

×

3922

        while (pIter2) {

×

3923

          size_t   keyLen2 = 0;

×

3924

          tb_uid_t uid = *(tb_uid_t*)taosHashGetKey(pIter2, &keyLen2);

×

3925

          SArray*  pTagList = *(SArray**)pIter2;

×

3927

          if (pVtbScan->genNewParam) {

×

3928

            code = buildAggOperatorParamForSingleChild(pInfo, uid, *groupid, pTagList, &pAggParam);

×

3929

            QUERY_CHECK_CODE(code, line, _return);

×

3930

            if (pAggParam) {

×

3931

              code = pAggOp->fpSet.getNextExtFn(pAggOp, pAggParam, pRes);

×

3932

              QUERY_CHECK_CODE(code, line, _return);

×

3934

              *pRes = NULL;

×

3937

            code = pAggOp->fpSet.getNextFn(pAggOp, pRes);

×

3938

            QUERY_CHECK_CODE(code, line, _return);

×

3941

          if (*pRes) {

×

3942

            pVtbScan->genNewParam = false;

×

3943

            (*pRes)->info.id.groupId = *groupid;

×

3944

            break;

×

3946

          pVtbScan->genNewParam = true;

×

3947

          pIter2 = taosHashIterate(vtbUidTagListMap, pIter2);

×

3948

          code = taosHashRemove(vtbUidTagListMap, &uid, keyLen);

×

3949

          QUERY_CHECK_CODE(code, line, _return);

×

3951

        if (*pRes) {

×

3952

          break;

×

3954

        pIter = taosHashIterate(pVtbScan->vtbGroupIdTagListMap, pIter);

×

3955

        code = taosHashRemove(pVtbScan->vtbGroupIdTagListMap, groupid, keyLen);

×

3956

        QUERY_CHECK_CODE(code, line, _return);

×

3961

    if (pInfo->vtbScan.batchProcessChild) {

×

3962

      code = buildAggOperatorParam(pInfo, &pAggParam);

×

3963

      QUERY_CHECK_CODE(code, line, _return);

×

3965

      code = pAggOp->fpSet.getNextExtFn(pAggOp, pAggParam, pRes);

×

3966

      QUERY_CHECK_CODE(code, line, _return);

×

3967

      setOperatorCompleted(pOperator);

×

3969

      void* pIter = taosHashIterate(pVtbScan->vtbUidTagListMap, NULL);

×

3970

      while (pIter) {

×

3971

        size_t   keyLen = 0;

×

3972

        tb_uid_t uid = *(tb_uid_t*)taosHashGetKey(pIter, &keyLen);

×

3973

        SArray*  pTagList = *(SArray**)pIter;

×

3975

        if (pVtbScan->genNewParam) {

×

3976

          code = buildAggOperatorParamForSingleChild(pInfo, uid, 0, pTagList, &pAggParam);

×

3977

          QUERY_CHECK_CODE(code, line, _return);

×

3979

          if (pAggParam) {

×

3980

            code = pAggOp->fpSet.getNextExtFn(pAggOp, pAggParam, pRes);

×

3981

            QUERY_CHECK_CODE(code, line, _return);

×

3983

            *pRes = NULL;

×

3986

          code = pAggOp->fpSet.getNextFn(pAggOp, pRes);

×

3987

          QUERY_CHECK_CODE(code, line, _return);

×

3990

        if (*pRes) {

×

3991

          pVtbScan->genNewParam = false;

×

3992

          break;

×

3994

        pVtbScan->genNewParam = true;

×

3995

        pIter = taosHashIterate(pVtbScan->vtbUidTagListMap, pIter);

×

3996

        code = taosHashRemove(pVtbScan->vtbUidTagListMap, &uid, keyLen);

×

3997

        QUERY_CHECK_CODE(code, line, _return);

×

4001

_return:

×

4002

  if (code) {

×

4003

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

×

4005

  return code;

×

4008

int32_t vtbAggNext(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

4009

  int32_t                    code = TSDB_CODE_SUCCESS;

×

4010

  int32_t                    line = 0;

×

4011

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

4012

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

4014

  QRY_PARAM_CHECK(pRes);

×

4015

  if (pOperator->status == OP_EXEC_DONE) {

×

4016

    return code;

×

4019

  code = pOperator->fpSet._openFn(pOperator);

×

4020

  QUERY_CHECK_CODE(code, line, _return);

×

4022

  if (pVtbScan->isSuperTable && taosArrayGetSize(pVtbScan->childTableList) == 0) {

×

4023

    setOperatorCompleted(pOperator);

×

4024

    return code;

×

4027

  code = virtualTableAggGetNext(pOperator, pRes);

×

4028

  QUERY_CHECK_CODE(code, line, _return);

×

4030

  return code;

×

4032

_return:

×

4033

  if (code) {

×

4034

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

×

4035

    pOperator->pTaskInfo->code = code;

×

4036

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

4038

  return code;

×

4049

static int32_t buildHashIntervalOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorParam** ppRes) {

×

4050

  int32_t                   code = TSDB_CODE_SUCCESS;

×

4051

  int32_t                   lino = 0;

×

4052

  SOperatorParam*           pParam = NULL;

×

4053

  SOperatorParam*           pExchangeParam = NULL;

×

4054

  SVtbScanDynCtrlInfo*      pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

4056

  pParam = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

4057

  QUERY_CHECK_NULL(pParam, code, lino, _return, terrno)

×

4059

  pParam->opType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL;

×

4060

  pParam->downstreamIdx = 0;

×

4061

  pParam->reUse = false;

×

4062

  pParam->pChildren = taosArrayInit(1, POINTER_BYTES);

×

4063

  QUERY_CHECK_NULL(pParam->pChildren, code, lino, _return, terrno)

×

4065

  pParam->value = taosMemoryMalloc(sizeof(SAggOperatorParam));

×

4066

  QUERY_CHECK_NULL(pParam->value, code, lino, _return, terrno)

×

4068

  code = buildBatchExchangeOperatorParamForVirtual(

×

4070

      (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MIN}, EX_SRC_TYPE_VSTB_INTERVAL_SCAN,

×

4072

  QUERY_CHECK_CODE(code, lino, _return);

×

4074

  QUERY_CHECK_NULL(taosArrayPush(pParam->pChildren, &pExchangeParam), code, lino, _return, terrno)

×

4075

  pExchangeParam = NULL;

×

4077

  *ppRes = pParam;

×

4079

  return code;

×

4080

_return:

×

4081

  if (pExchangeParam) {

×

4082

    freeOperatorParam(pExchangeParam, OP_GET_PARAM);

×

4084

  if (pParam) {

×

4085

    freeOperatorParam(pParam, OP_GET_PARAM);

×

4087

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

×

4088

  return code;

×

4100

static int32_t buildSplitIntervalMergeOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorInfo* pMergeOp,

×

4102

  int32_t              code = TSDB_CODE_SUCCESS;

×

4103

  int32_t              lino = 0;

×

4104

  SVtbScanDynCtrlInfo* pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

4105

  SOperatorParam*      pExchangeParam = NULL;

×

4106

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

4107

  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno)

×

4109

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_MERGE;

×

4110

  (*ppRes)->downstreamIdx = 0;

×

4111

  (*ppRes)->reUse = false;

×

4112

  (*ppRes)->value = taosMemoryCalloc(1, sizeof(SMergeOperatorParam));

×

4113

  QUERY_CHECK_NULL((*ppRes)->value, code, lino, _return, terrno)

×

4115

  (*ppRes)->pChildren = taosArrayInit(pMergeOp->numOfDownstream, POINTER_BYTES);

×

4116

  QUERY_CHECK_NULL((*ppRes)->pChildren, code, lino, _return, terrno)

×

4118

  for (int32_t i = 0; i < pMergeOp->numOfDownstream; ++i) {

×

4119

    code = buildBatchExchangeOperatorParamForVirtual(

×

4121

        (STimeWindow){.skey = INT64_MAX, .ekey = INT64_MIN}, EX_SRC_TYPE_VSTB_PART_INTERVAL_SCAN,

×

4123

    QUERY_CHECK_CODE(code, lino, _return);

×

4124

    QUERY_CHECK_NULL(taosArrayPush((*ppRes)->pChildren, &pExchangeParam), code, lino, _return, terrno)

×

4125

    pExchangeParam = NULL;

×

4128

  return code;

×

4130

_return:

×

4131

  if (pExchangeParam) {

×

4132

    freeOperatorParam(pExchangeParam, OP_GET_PARAM);

×

4134

  freeOperatorParam(*ppRes, OP_GET_PARAM);

×

4135

  *ppRes = NULL;

×

4136

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

×

4137

  return code;

×

4149

static int32_t buildSplitIntervalOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorInfo* pIntervalOp,

×

4151

  int32_t         code = TSDB_CODE_SUCCESS;

×

4152

  int32_t         lino = 0;

×

4153

  SOperatorInfo*  pMergeOp = NULL;

×

4154

  SOperatorParam* pMergeParam = NULL;

×

4156

  QUERY_CHECK_NULL(pIntervalOp->pDownstream, code, lino, _return, TSDB_CODE_INVALID_PARA)

×

4157

  pMergeOp = pIntervalOp->pDownstream[0];

×

4158

  QUERY_CHECK_NULL(pMergeOp, code, lino, _return, TSDB_CODE_INVALID_PARA)

×

4159

  if (QUERY_NODE_PHYSICAL_PLAN_MERGE != pMergeOp->operatorType) {

×

4160

    qError("%s invalid downstream operator type %d for split interval", __func__, pMergeOp->operatorType);

×

4161

    return TSDB_CODE_INVALID_PARA;

×

4164

  *ppRes = taosMemoryCalloc(1, sizeof(SOperatorParam));

×

4165

  QUERY_CHECK_NULL(*ppRes, code, lino, _return, terrno)

×

4167

  (*ppRes)->opType = QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL;

×

4168

  (*ppRes)->downstreamIdx = 0;

×

4169

  (*ppRes)->reUse = false;

×

4170

  (*ppRes)->value = taosMemoryCalloc(1, sizeof(SAggOperatorParam));

×

4171

  QUERY_CHECK_NULL((*ppRes)->value, code, lino, _return, terrno)

×

4173

  (*ppRes)->pChildren = taosArrayInit(1, POINTER_BYTES);

×

4174

  QUERY_CHECK_NULL((*ppRes)->pChildren, code, lino, _return, terrno)

×

4176

  code = buildSplitIntervalMergeOperatorParam(pInfo, pMergeOp, &pMergeParam);

×

4177

  QUERY_CHECK_CODE(code, lino, _return);

×

4178

  QUERY_CHECK_NULL(taosArrayPush((*ppRes)->pChildren, &pMergeParam), code, lino, _return, terrno)

×

4179

  pMergeParam = NULL;

×

4181

  return code;

×

4183

_return:

×

4184

  if (pMergeParam) {

×

4185

    freeOperatorParam(pMergeParam, OP_GET_PARAM);

×

4187

  freeOperatorParam(*ppRes, OP_GET_PARAM);

×

4188

  *ppRes = NULL;

×

4189

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

×

4190

  return code;

×

4202

static int32_t buildIntervalOperatorParam(SDynQueryCtrlOperatorInfo* pInfo, SOperatorInfo* pIntervalOp,

×

4204

  if (NULL == pIntervalOp) {

×

4205

    return TSDB_CODE_INVALID_PARA;

×

4208

  switch (pIntervalOp->operatorType) {

×

4209

    case QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL:

×

4210

      return buildHashIntervalOperatorParam(pInfo, ppRes);

×

4211

    case QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL:

×

4212

      return buildSplitIntervalOperatorParam(pInfo, pIntervalOp, ppRes);

×

4213

    default:

×

4214

      qError("%s unsupported interval operator type %d", __func__, pIntervalOp->operatorType);

×

4215

      return TSDB_CODE_INVALID_PARA;

×

4227

int32_t vtbIntervalNext(SOperatorInfo* pOperator, SSDataBlock** pRes) {

×

4228

  int32_t                    code = TSDB_CODE_SUCCESS;

×

4229

  int32_t                    line = 0;

×

4230

  SDynQueryCtrlOperatorInfo* pInfo = pOperator->info;

×

4231

  SVtbScanDynCtrlInfo*       pVtbScan = (SVtbScanDynCtrlInfo*)&pInfo->vtbScan;

×

4232

  SOperatorInfo*             pInterval = pOperator->pDownstream[pOperator->numOfDownstream - 1];

×

4233

  SOperatorParam*            pIntervalParam = NULL;

×

4234

  SSDataBlock*               pResult = NULL;

×

4236

  QRY_PARAM_CHECK(pRes);

×

4237

  if (pOperator->status == OP_EXEC_DONE) {

×

4238

    return code;

×

4241

  code = pOperator->fpSet._openFn(pOperator);

×

4242

  QUERY_CHECK_CODE(code, line, _return);

×

4244

  if (pVtbScan->genNewParam) {

×

4245

    qDebug("%s vtb interval split start, intervalOp:%s type:%d", GET_TASKID(pOperator->pTaskInfo), pInterval->name,

×

4247

    code = buildIntervalOperatorParam(pInfo, pInterval, &pIntervalParam);

×

4248

    QUERY_CHECK_CODE(code, line, _return);

×

4250

    code = pInterval->fpSet.getNextExtFn(pInterval, pIntervalParam, &pResult);

×

4251

    QUERY_CHECK_CODE(code, line, _return);

×

4253

    pVtbScan->genNewParam = false;

×

4255

    code = pInterval->fpSet.getNextFn(pInterval, &pResult);

×

4256

    QUERY_CHECK_CODE(code, line, _return);

×

4259

  if (pResult) {

×

4260

    qDebug("%s vtb interval got result rows:%" PRId64 " status:%d", GET_TASKID(pOperator->pTaskInfo),

×

4262

    *pRes = pResult;

×

4264

    qDebug("%s vtb interval got empty result, interval status:%d", GET_TASKID(pOperator->pTaskInfo), pInterval->status);

×

4265

    *pRes = NULL;

×

4266

    setOperatorCompleted(pOperator);

×

4269

  return code;

×

4271

_return:

×

4272

  if (code) {

×

4273

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

×

4274

    pOperator->pTaskInfo->code = code;

×

4275

    T_LONG_JMP(pOperator->pTaskInfo->env, code);

×

4277

  return code;

×

4280

int32_t createDynQueryCtrlOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream,

×

4283

  QRY_PARAM_CHECK(pOptrInfo);

×

4285

  int32_t                    code = TSDB_CODE_SUCCESS;

×

4286

  int32_t                    line = 0;

×

4287

  __optr_fn_t                nextFp = NULL;

×

4288

  __optr_open_fn_t           openFp = NULL;

×

4289

  SOperatorInfo*             pOperator = NULL;

×

4290

  SDynQueryCtrlOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SDynQueryCtrlOperatorInfo));

×

4291

  QUERY_CHECK_NULL(pInfo, code, line, _error, terrno)

×

4293

  pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

4294

  QUERY_CHECK_NULL(pOperator, code, line, _error, terrno)

×

4296

  pOperator->pPhyNode = pPhyciNode;

×

4297

  pTaskInfo->dynamicTask = (int8_t)pPhyciNode->node.dynamicOp;

×

4299

  code = appendDownstream(pOperator, pDownstream, numOfDownstream);

×

4300

  QUERY_CHECK_CODE(code, line, _error);

×

4302

  setOperatorInfo(pOperator, "DynQueryCtrlOperator", QUERY_NODE_PHYSICAL_PLAN_DYN_QUERY_CTRL, false, OP_NOT_OPENED,

×

4305

  pInfo->qType = pPhyciNode->qType;

×

4306

  switch (pInfo->qType) {

×

4307

    case DYN_QTYPE_STB_HASH:

×

4308

      TAOS_MEMCPY(&pInfo->stbJoin.basic, &pPhyciNode->stbJoin, sizeof(pPhyciNode->stbJoin));

×

4309

      pInfo->stbJoin.pOutputDataBlockDesc = pPhyciNode->node.pOutputDataBlockDesc;

×

4310

      code = initSeqStbJoinTableHash(&pInfo->stbJoin.ctx.prev, pInfo->stbJoin.basic.batchFetch);

×

4311

      QUERY_CHECK_CODE(code, line, _error);

×

4312

      nextFp = seqStableJoin;

×

4313

      openFp = optrDummyOpenFn;

×

4314

      break;

×

4315

    case DYN_QTYPE_VTB_SCAN:

×

4316

      code = initVtbScanInfo(pInfo, pMsgCb, pPhyciNode, pTaskInfo);

×

4317

      QUERY_CHECK_CODE(code, line, _error);

×

4318

      nextFp = vtbScanNext;

×

4319

      openFp = vtbScanOpen;

×

4320

      break;

×

4321

    case DYN_QTYPE_VTB_TS_SCAN:

×

4322

      code = initVtbScanInfo(pInfo, pMsgCb, pPhyciNode, pTaskInfo);

×

4323

      QUERY_CHECK_CODE(code, line, _error);

×

4324

      nextFp = vtbTsScanNext;

×

4325

      openFp = vtbDefaultOpen;

×

4326

      break;

×

4327

    case DYN_QTYPE_VTB_WINDOW:

×

4328

      code = initVtbScanInfo(pInfo, pMsgCb, pPhyciNode, pTaskInfo);

×

4329

      QUERY_CHECK_CODE(code, line, _error);

×

4330

      code = initVtbWindowInfo(pInfo, pPhyciNode, pTaskInfo, pOperator);

×

4331

      QUERY_CHECK_CODE(code, line, _error);

×

4332

      nextFp = vtbWindowNext;

×

4333

      openFp = vtbWindowOpen;

×

4334

      break;

×

4335

    case DYN_QTYPE_VTB_AGG:

×

4336

      code = initVtbScanInfo(pInfo, pMsgCb, pPhyciNode, pTaskInfo);

×

4337

      QUERY_CHECK_CODE(code, line, _error);

×

4338

      nextFp = vtbAggNext;

×

4339

      openFp = vtbDefaultOpen;

×

4340

      break;

×

4341

    case DYN_QTYPE_VTB_INTERVAL:

×

4342

      code = initVtbScanInfo(pInfo, pMsgCb, pPhyciNode, pTaskInfo);

×

4343

      QUERY_CHECK_CODE(code, line, _error);

×

4344

      nextFp = vtbIntervalNext;

×

4345

      openFp = vtbDefaultOpen;

×

4346

      break;

×

4347

    default:

×

4348

      qError("unsupported dynamic query ctrl type: %d", pInfo->qType);

×

4349

      code = TSDB_CODE_INVALID_PARA;

×

4350

      goto _error;

×

4353

  pOperator->fpSet = createOperatorFpSet(openFp, nextFp, NULL, destroyDynQueryCtrlOperator, optrDefaultBufFn,

×

4356

  setOperatorResetStateFn(pOperator, resetDynQueryCtrlOperState);

×

4357

  *pOptrInfo = pOperator;

×

4358

  return TSDB_CODE_SUCCESS;

×

4360

_error:

×

4361

  if (pInfo != NULL) {

×

4362

    destroyDynQueryCtrlOperator(pInfo);

×

4364

  qError("failed to create dyn query ctrl operator, %s code:%s, line:%d", __func__, tstrerror(code), line);

×

4365

  destroyOperatorAndDownstreams(pOperator, pDownstream, numOfDownstream);

×

4366

  pTaskInfo->code = code;

×

4367

  return code;

×

taosdata / TDengine / #4986

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

taosdata / TDengine / #4986

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