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

UNCOV

bool hJoinBlkReachThreshold(SHJoinOperatorInfo* pInfo, int64_t blkRows) {

×

UNCOV

33

  if (INT64_MAX == pInfo->ctx.limit || pInfo->pFinFilter != NULL) {

×

UNCOV

34

    return blkRows >= pInfo->blkThreshold;

×

37

  return (pInfo->execInfo.resRows + blkRows) >= pInfo->ctx.limit;

×

40

int32_t hJoinHandleMidRemains(SHJoinOperatorInfo* pJoin, SHJoinCtx* pCtx) {

×

41

  TSWAP(pJoin->midBlk, pJoin->finBlk);

×

43

  pCtx->midRemains = false;

×

45

  return TSDB_CODE_SUCCESS;

×

48

int32_t hJoinCopyMergeMidBlk(SHJoinCtx* pCtx, SSDataBlock** ppMid, SSDataBlock** ppFin) {

×

49

  SSDataBlock* pLess = *ppMid;

×

50

  SSDataBlock* pMore = *ppFin;

×

62

  int32_t totalRows = pMore->info.rows + pLess->info.rows;

×

63

  if (totalRows <= pMore->info.capacity) {

×

64

    HJ_ERR_RET(blockDataMerge(pMore, pLess));

×

65

    blockDataCleanup(pLess);

×

66

    pCtx->midRemains = false;

×

68

    int32_t copyRows = pMore->info.capacity - pMore->info.rows;

×

69

    HJ_ERR_RET(blockDataMergeNRows(pMore, pLess, pLess->info.rows - copyRows, copyRows));

×

70

    blockDataShrinkNRows(pLess, copyRows);

×

71

    pCtx->midRemains = true;

×

80

  return TSDB_CODE_SUCCESS;

×

UNCOV

83

int32_t hJoinSetImplFp(SHJoinOperatorInfo* pJoin) {

×

UNCOV

84

  switch (pJoin->joinType) {

×

UNCOV

85

    case JOIN_TYPE_INNER:

×

UNCOV

86

      pJoin->joinFp = hInnerJoinDo;

×

UNCOV

87

      break;

×

88

    case JOIN_TYPE_LEFT:

×

90

      switch (pJoin->subType) {

×

91

        case JOIN_STYPE_OUTER:

×

93

          break;

×

94

        default:

×

95

          break;

×

97

      break;

×

99

    default:

×

100

      break;

×

UNCOV

103

  return TSDB_CODE_SUCCESS;

×

UNCOV

107

int32_t hJoinLaunchPrimExpr(SSDataBlock* pBlock, SHJoinTableCtx* pTable, int32_t startIdx, int32_t endIdx) {

×

UNCOV

108

  if (NULL == pTable->primExpr) {

×

UNCOV

109

    return TSDB_CODE_SUCCESS;

×

112

  SHJoinPrimExprCtx* pCtx = &pTable->primCtx;

×

113

  SColumnInfoData* pPrimIn = taosArrayGet(pBlock->pDataBlock, pTable->primCol->srcSlot);

×

114

  SColumnInfoData* pPrimOut = taosArrayGet(pBlock->pDataBlock, pTable->primCtx.targetSlotId);

×

115

  if (0 != pCtx->timezoneUnit) {

×

116

    for (int32_t i = startIdx; i <= endIdx; ++i) {

×

117

      ((int64_t*)pPrimOut->pData)[i] = ((int64_t*)pPrimIn->pData)[i] - (((int64_t*)pPrimIn->pData)[i] + pCtx->timezoneUnit) % pCtx->truncateUnit;

×

120

    for (int32_t i = startIdx; i <= endIdx; ++i) {

×

121

      ((int64_t*)pPrimOut->pData)[i] = ((int64_t*)pPrimIn->pData)[i] / pCtx->truncateUnit * pCtx->truncateUnit;

×

125

  return TSDB_CODE_SUCCESS;

×

129

static int64_t hJoinGetSingleKeyRowsNum(SBufRowInfo* pRow) {

×

130

  int64_t rows = 0;

×

131

  while (pRow) {

×

132

    rows++;

×

133

    pRow = pRow->next;

×

135

  return rows;

×

138

static int64_t hJoinGetRowsNumOfKeyHash(SSHashObj* pHash) {

×

139

  SGroupData* pGroup = NULL;

×

140

  int32_t iter = 0;

×

141

  int64_t rowsNum = 0;

×

143

  while (NULL != (pGroup = tSimpleHashIterate(pHash, pGroup, &iter))) {

×

144

    int32_t* pKey = tSimpleHashGetKey(pGroup, NULL);

×

145

    int64_t rows = hJoinGetSingleKeyRowsNum(pGroup->rows);

×

147

    rowsNum += rows;

×

150

  return rowsNum;

×

UNCOV

153

static int32_t hJoinInitKeyColsInfo(SHJoinTableCtx* pTable, SNodeList* pList) {

×

UNCOV

154

  pTable->keyNum = LIST_LENGTH(pList);

×

UNCOV

156

  pTable->keyCols = taosMemoryMalloc(pTable->keyNum * sizeof(SHJoinColInfo));

×

UNCOV

157

  if (NULL == pTable->keyCols) {

×

158

    return terrno;

×

UNCOV

161

  int64_t bufSize = 0;

×

UNCOV

162

  int32_t i = 0;

×

UNCOV

163

  SNode* pNode = NULL;

×

UNCOV

164

  FOREACH(pNode, pList) {

×

UNCOV

165

    SColumnNode* pColNode = (SColumnNode*)pNode;

×

UNCOV

166

    pTable->keyCols[i].srcSlot = pColNode->slotId;

×

UNCOV

167

    pTable->keyCols[i].vardata = IS_VAR_DATA_TYPE(pColNode->node.resType.type);

×

UNCOV

168

    pTable->keyCols[i].bytes = pColNode->node.resType.bytes;

×

UNCOV

169

    bufSize += pColNode->node.resType.bytes;

×

UNCOV

170

    ++i;

×

UNCOV

173

  if (pTable->keyNum > 1) {

×

UNCOV

174

    pTable->keyBuf = taosMemoryMalloc(bufSize);

×

UNCOV

175

    if (NULL == pTable->keyBuf) {

×

176

      return terrno;

×

UNCOV

180

  return TSDB_CODE_SUCCESS;

×

UNCOV

183

static void hJoinGetValColsNum(SNodeList* pList, int32_t blkId, int32_t* colNum) {

×

UNCOV

184

  *colNum = 0;

×

UNCOV

186

  SNode* pNode = NULL;

×

UNCOV

187

  FOREACH(pNode, pList) {

×

UNCOV

188

    STargetNode* pTarget = (STargetNode*)pNode;

×

UNCOV

189

    SColumnNode* pCol = (SColumnNode*)pTarget->pExpr;

×

UNCOV

190

    if (pCol->dataBlockId == blkId) {

×

UNCOV

191

      (*colNum)++;

×

UNCOV

194

×

UNCOV

196

static bool hJoinIsValColInKeyCols(int16_t slotId, int32_t keyNum, SHJoinColInfo* pKeys, int32_t* pKeyIdx) {

×

UNCOV

197

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

×

UNCOV

198

    if (pKeys[i].srcSlot == slotId) {

×

199

      *pKeyIdx = i;

×

200

      return true;

×

UNCOV

204

  return false;

×

UNCOV

207

static int32_t hJoinInitValColsInfo(SHJoinTableCtx* pTable, SNodeList* pList) {

×

UNCOV

208

  hJoinGetValColsNum(pList, pTable->blkId, &pTable->valNum);

×

UNCOV

209

  if (pTable->valNum == 0) {

×

210

    return TSDB_CODE_SUCCESS;

×

UNCOV

213

  pTable->valCols = taosMemoryMalloc(pTable->valNum * sizeof(SHJoinColInfo));

×

UNCOV

214

  if (NULL == pTable->valCols) {

×

215

    return terrno;

×

UNCOV

218

  int32_t i = 0;

×

UNCOV

219

  int32_t colNum = 0;

×

UNCOV

220

  SNode* pNode = NULL;

×

UNCOV

221

  FOREACH(pNode, pList) {

×

UNCOV

222

    STargetNode* pTarget = (STargetNode*)pNode;

×

UNCOV

223

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

×

UNCOV

224

    if (pColNode->dataBlockId == pTable->blkId) {

×

UNCOV

225

      if (hJoinIsValColInKeyCols(pColNode->slotId, pTable->keyNum, pTable->keyCols, &pTable->valCols[i].srcSlot)) {

×

226

        pTable->valCols[i].keyCol = true;

×

UNCOV

228

        pTable->valCols[i].keyCol = false;

×

UNCOV

229

        pTable->valCols[i].srcSlot = pColNode->slotId;

×

UNCOV

230

        pTable->valColExist = true;

×

UNCOV

231

        colNum++;

×

UNCOV

233

      pTable->valCols[i].dstSlot = pTarget->slotId;

×

UNCOV

234

      pTable->valCols[i].vardata = IS_VAR_DATA_TYPE(pColNode->node.resType.type);

×

UNCOV

235

      if (pTable->valCols[i].vardata && !pTable->valCols[i].keyCol) {

×

236

        if (NULL == pTable->valVarCols) {

×

237

          pTable->valVarCols = taosArrayInit(pTable->valNum, sizeof(int32_t));

×

238

          if (NULL == pTable->valVarCols) {

×

239

            return terrno;

×

242

        if (NULL == taosArrayPush(pTable->valVarCols, &i)) {

×

243

          return terrno;

×

UNCOV

246

      pTable->valCols[i].bytes = pColNode->node.resType.bytes;

×

UNCOV

247

      if (!pTable->valCols[i].keyCol && !pTable->valCols[i].vardata) {

×

UNCOV

248

        pTable->valBufSize += pColNode->node.resType.bytes;

×

UNCOV

250

      i++;

×

UNCOV

254

  pTable->valBitMapSize = BitmapLen(colNum);

×

UNCOV

255

  pTable->valBufSize += pTable->valBitMapSize;

×

UNCOV

257

  return TSDB_CODE_SUCCESS;

×

UNCOV

260

static int32_t hJoinInitPrimKeyInfo(SHJoinTableCtx* pTable, int32_t slotId) {

×

UNCOV

261

  pTable->primCol = taosMemoryMalloc(sizeof(SHJoinColMap));

×

UNCOV

262

  if (NULL == pTable->primCol) {

×

263

    return terrno;

×

UNCOV

266

  pTable->primCol->srcSlot = slotId;

×

UNCOV

268

  return TSDB_CODE_SUCCESS;

×

UNCOV

272

static int32_t hJoinInitPrimExprCtx(SNode* pNode, SHJoinPrimExprCtx* pCtx, SHJoinTableCtx* pTable) {

×

UNCOV

273

  if (NULL == pNode) {

×

UNCOV

274

    pCtx->targetSlotId = pTable->primCol->srcSlot;

×

UNCOV

275

    return TSDB_CODE_SUCCESS;

×

278

  if (QUERY_NODE_TARGET != nodeType(pNode)) {

×

279

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

282

  STargetNode* pTarget = (STargetNode*)pNode;

×

283

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

×

284

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

287

  SFunctionNode* pFunc = (SFunctionNode*)pTarget->pExpr;

×

288

  if (FUNCTION_TYPE_TIMETRUNCATE != pFunc->funcType) {

×

289

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

292

  if (4 != pFunc->pParameterList->length && 5 != pFunc->pParameterList->length) {

×

293

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

296

  SValueNode* pUnit = (SValueNode*)nodesListGetNode(pFunc->pParameterList, 1);

×

297

  SValueNode* pCurrTz = (5 == pFunc->pParameterList->length) ? (SValueNode*)nodesListGetNode(pFunc->pParameterList, 2) : NULL;

×

298

  SValueNode* pTimeZone = (5 == pFunc->pParameterList->length) ? (SValueNode*)nodesListGetNode(pFunc->pParameterList, 4) : (SValueNode*)nodesListGetNode(pFunc->pParameterList, 3);

×

300

  pCtx->truncateUnit = pUnit->typeData;

×

301

  if ((NULL == pCurrTz || 1 == pCurrTz->typeData) && pCtx->truncateUnit >= (86400 * TSDB_TICK_PER_SECOND(pFunc->node.resType.precision))) {

×

302

    pCtx->timezoneUnit = offsetFromTz(varDataVal(pTimeZone->datum.p), TSDB_TICK_PER_SECOND(pFunc->node.resType.precision));

×

305

  pCtx->targetSlotId = pTarget->slotId;

×

307

  return TSDB_CODE_SUCCESS;

×

UNCOV

311

static int32_t hJoinInitTableInfo(SHJoinOperatorInfo* pJoin, SHashJoinPhysiNode* pJoinNode, SOperatorInfo** pDownstream, int32_t idx, SQueryStat* pStat) {

×

UNCOV

312

  SNodeList* pKeyList = NULL;

×

UNCOV

313

  SHJoinTableCtx* pTable = &pJoin->tbs[idx];

×

UNCOV

314

  pTable->downStream = pDownstream[idx];

×

UNCOV

315

  pTable->blkId = pDownstream[idx]->resultDataBlockId;

×

UNCOV

316

  if (0 == idx) {

×

UNCOV

317

    pKeyList = pJoinNode->pOnLeft;

×

UNCOV

318

    pTable->hasTimeRange = pJoinNode->timeRangeTarget & 0x1;

×

UNCOV

320

    pKeyList = pJoinNode->pOnRight;

×

UNCOV

321

    pTable->hasTimeRange = pJoinNode->timeRangeTarget & 0x2;

×

UNCOV

324

  HJ_ERR_RET(hJoinInitPrimKeyInfo(pTable, (0 == idx) ? pJoinNode->leftPrimSlotId : pJoinNode->rightPrimSlotId));

×

UNCOV

326

  int32_t code = hJoinInitKeyColsInfo(pTable, pKeyList);

×

UNCOV

327

  if (code) {

×

328

    return code;

×

UNCOV

330

  code = hJoinInitValColsInfo(pTable, pJoinNode->pTargets);

×

UNCOV

331

  if (code) {

×

332

    return code;

×

UNCOV

335

  TAOS_MEMCPY(&pTable->inputStat, pStat, sizeof(*pStat));

×

UNCOV

337

  HJ_ERR_RET(hJoinInitPrimExprCtx(pTable->primExpr, &pTable->primCtx, pTable));

×

UNCOV

339

  return TSDB_CODE_SUCCESS;

×

UNCOV

342

static void hJoinSetBuildAndProbeTable(SHJoinOperatorInfo* pInfo, SHashJoinPhysiNode* pJoinNode) {

×

UNCOV

343

  int32_t buildIdx = 0;

×

UNCOV

344

  int32_t probeIdx = 1;

×

UNCOV

346

  pInfo->joinType = pJoinNode->joinType;

×

UNCOV

347

  pInfo->subType = pJoinNode->subType;

×

UNCOV

349

  switch (pInfo->joinType) {

×

UNCOV

350

    case JOIN_TYPE_INNER:

×

UNCOV

352

      if (pInfo->tbs[0].inputStat.inputRowNum <= pInfo->tbs[1].inputStat.inputRowNum) {

×

UNCOV

353

        buildIdx = 0;

×

UNCOV

354

        probeIdx = 1;

×

356

        buildIdx = 1;

×

357

        probeIdx = 0;

×

UNCOV

359

      break;

×

360

    case JOIN_TYPE_LEFT:

×

361

      buildIdx = 1;

×

362

      probeIdx = 0;

×

363

      break;

×

364

    case JOIN_TYPE_RIGHT:

×

365

      buildIdx = 0;

×

366

      probeIdx = 1;

×

367

      break;

×

368

    default:

×

369

      break;

×

UNCOV

372

  pInfo->pBuild = &pInfo->tbs[buildIdx];

×

UNCOV

373

  pInfo->pProbe = &pInfo->tbs[probeIdx];

×

UNCOV

375

  pInfo->pBuild->downStreamIdx = buildIdx;

×

UNCOV

376

  pInfo->pProbe->downStreamIdx = probeIdx;

×

UNCOV

378

  if (0 == buildIdx) {

×

UNCOV

379

    pInfo->pBuild->primExpr = pJoinNode->leftPrimExpr;

×

UNCOV

380

    pInfo->pProbe->primExpr = pJoinNode->rightPrimExpr;

×

382

    pInfo->pBuild->primExpr = pJoinNode->rightPrimExpr;

×

383

    pInfo->pProbe->primExpr = pJoinNode->leftPrimExpr;

×

UNCOV

385

×

UNCOV

387

static int32_t hJoinBuildResColsMap(SHJoinOperatorInfo* pInfo, SHashJoinPhysiNode* pJoinNode) {

×

UNCOV

388

  pInfo->pResColNum = pJoinNode->pTargets->length;

×

UNCOV

389

  pInfo->pResColMap = taosMemoryCalloc(pJoinNode->pTargets->length, sizeof(int8_t));

×

UNCOV

390

  if (NULL == pInfo->pResColMap) {

×

391

    return terrno;

×

UNCOV

394

  SNode* pNode = NULL;

×

UNCOV

395

  int32_t i = 0;

×

UNCOV

396

  FOREACH(pNode, pJoinNode->pTargets) {

×

UNCOV

397

    STargetNode* pTarget = (STargetNode*)pNode;

×

UNCOV

398

    SColumnNode* pCol = (SColumnNode*)pTarget->pExpr;

×

UNCOV

399

    if (pCol->dataBlockId == pInfo->pBuild->blkId) {

×

UNCOV

400

      pInfo->pResColMap[i] = 1;

×

UNCOV

403

    i++;

×

UNCOV

406

  return TSDB_CODE_SUCCESS;

×

UNCOV

412

  page.pageSize = HASH_JOIN_DEFAULT_PAGE_SIZE;

×

UNCOV

413

  page.offset = 0;

×

UNCOV

414

  page.data = taosMemoryMalloc(page.pageSize);

×

UNCOV

415

  if (NULL == page.data) {

×

416

    return terrno;

×

UNCOV

419

  if (NULL == taosArrayPush(pRowBufs, &page)) {

×

420

    return terrno;

×

UNCOV

422

  return TSDB_CODE_SUCCESS;

×

UNCOV

425

static int32_t hJoinInitBufPages(SHJoinOperatorInfo* pInfo) {

×

UNCOV

426

  pInfo->pRowBufs = taosArrayInit(32, sizeof(SBufPageInfo));

×

UNCOV

427

  if (NULL == pInfo->pRowBufs) {

×

428

    return terrno;

×

UNCOV

431

  return hJoinAddPageToBufs(pInfo->pRowBufs);

×

UNCOV

434

static void hJoinFreeTableInfo(SHJoinTableCtx* pTable) {

×

UNCOV

435

  taosMemoryFreeClear(pTable->keyCols);

×

UNCOV

436

  taosMemoryFreeClear(pTable->keyBuf);

×

UNCOV

437

  taosMemoryFreeClear(pTable->valCols);

×

UNCOV

438

  taosArrayDestroy(pTable->valVarCols);

×

UNCOV

439

  taosMemoryFree(pTable->primCol);

×

UNCOV

440

×

UNCOV

442

static void hJoinFreeBufPage(void* param) {

×

UNCOV

443

  SBufPageInfo* pInfo = (SBufPageInfo*)param;

×

UNCOV

444

  taosMemoryFree(pInfo->data);

×

UNCOV

445

×

UNCOV

447

static void hJoinDestroyKeyHash(SSHashObj** ppHash) {

×

UNCOV

448

  if (NULL == ppHash || NULL == (*ppHash)) {

×

UNCOV

449

    return;

×

UNCOV

452

  void*   pIte = NULL;

×

UNCOV

453

  int32_t iter = 0;

×

UNCOV

454

  while ((pIte = tSimpleHashIterate(*ppHash, pIte, &iter)) != NULL) {

×

UNCOV

455

    SGroupData* pGroup = pIte;

×

UNCOV

456

    SBufRowInfo* pRow = pGroup->rows;

×

UNCOV

457

    SBufRowInfo* pNext = NULL;

×

UNCOV

458

    while (pRow) {

×

UNCOV

459

      pNext = pRow->next;

×

UNCOV

460

      taosMemoryFree(pRow);

×

UNCOV

461

      pRow = pNext;

×

UNCOV

465

  tSimpleHashCleanup(*ppHash);

×

UNCOV

466

  *ppHash = NULL;

×

UNCOV

470

  *ppData = NULL;

×

UNCOV

472

  if ((uint16_t)-1 == pRow->pageId) {

×

473

    return TSDB_CODE_SUCCESS;

×

UNCOV

475

  SBufPageInfo *pPage = taosArrayGet(pRowBufs, pRow->pageId);

×

UNCOV

476

  if (NULL == pPage) {

×

477

    qError("fail to get %d page, total:%d", pRow->pageId, (int32_t)taosArrayGetSize(pRowBufs));

×

478

    QRY_ERR_RET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

481

  *ppData = pPage->data + pRow->offset;

×

UNCOV

483

  return TSDB_CODE_SUCCESS;

×

UNCOV

486

static int32_t hJoinCopyResRowsToBlock(SHJoinOperatorInfo* pJoin, int32_t rowNum, SBufRowInfo* pStart, SSDataBlock* pRes) {

×

UNCOV

487

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

UNCOV

488

  SHJoinTableCtx* pProbe = pJoin->pProbe;

×

UNCOV

489

  int32_t buildIdx = 0, buildValIdx = 0;

×

UNCOV

490

  int32_t probeIdx = 0;

×

UNCOV

491

  SBufRowInfo* pRow = pStart;

×

UNCOV

492

  int32_t code = 0;

×

UNCOV

493

  char* pData = NULL;

×

UNCOV

495

  for (int32_t r = 0; r < rowNum; ++r) {

×

UNCOV

496

    HJ_ERR_RET(hJoinRetrieveColDataFromRowBufs(pJoin->pRowBufs, pRow, &pData));

×

UNCOV

498

    char* pValData = pData + pBuild->valBitMapSize;

×

UNCOV

499

    char* pKeyData = pProbe->keyData;

×

UNCOV

500

    buildIdx = buildValIdx = probeIdx = 0;

×

UNCOV

501

    for (int32_t i = 0; i < pJoin->pResColNum; ++i) {

×

UNCOV

502

      if (pJoin->pResColMap[i]) {

×

UNCOV

503

        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pBuild->valCols[buildIdx].dstSlot);

×

UNCOV

504

        if (pBuild->valCols[buildIdx].keyCol) {

×

505

          code = colDataSetVal(pDst, pRes->info.rows + r, pKeyData, false);

×

506

          if (code) {

×

507

            return code;

×

509

          pKeyData += pBuild->valCols[buildIdx].vardata ? varDataTLen(pKeyData) : pBuild->valCols[buildIdx].bytes;

×

UNCOV

511

          if (colDataIsNull_f(pData, buildValIdx)) {

×

512

            code = colDataSetVal(pDst, pRes->info.rows + r, NULL, true);

×

513

            if (code) {

×

514

              return code;

×

UNCOV

517

            code = colDataSetVal(pDst, pRes->info.rows + r, pValData, false);

×

UNCOV

518

            if (code) {

×

519

              return code;

×

UNCOV

521

            pValData += pBuild->valCols[buildIdx].vardata ? varDataTLen(pValData) : pBuild->valCols[buildIdx].bytes;

×

UNCOV

523

          buildValIdx++;

×

UNCOV

525

        buildIdx++;

×

UNCOV

526

      } else if (0 == r) {

×

UNCOV

527

        SColumnInfoData* pSrc = taosArrayGet(pJoin->ctx.pProbeData->pDataBlock, pProbe->valCols[probeIdx].srcSlot);

×

UNCOV

528

        SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pProbe->valCols[probeIdx].dstSlot);

×

UNCOV

530

        code = colDataCopyNItems(pDst, pRes->info.rows, colDataGetData(pSrc, pJoin->ctx.probeStartIdx), rowNum, colDataIsNull_s(pSrc, pJoin->ctx.probeStartIdx));

×

UNCOV

531

        if (code) {

×

532

          return code;

×

UNCOV

534

        probeIdx++;

×

UNCOV

537

    pRow = pRow->next;

×

UNCOV

540

  return TSDB_CODE_SUCCESS;

×

543

int32_t hJoinCopyNMatchRowsToBlock(SHJoinOperatorInfo* pJoin, SSDataBlock* pRes, int32_t startIdx, int32_t rows) {

×

544

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

545

  SHJoinTableCtx* pProbe = pJoin->pProbe;

×

546

  int32_t buildIdx = 0;

×

547

  int32_t probeIdx = 0;

×

548

  int32_t code = 0;

×

550

  for (int32_t i = 0; i < pJoin->pResColNum; ++i) {

×

551

    if (pJoin->pResColMap[i]) {

×

552

      SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pBuild->valCols[buildIdx].dstSlot);

×

553

      colDataSetNItemsNull(pDst, pRes->info.rows, rows);

×

555

      buildIdx++;

×

557

      SColumnInfoData* pSrc = taosArrayGet(pJoin->ctx.pProbeData->pDataBlock, pProbe->valCols[probeIdx].srcSlot);

×

558

      SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, pProbe->valCols[probeIdx].dstSlot);

×

560

      QRY_ERR_RET(colDataAssignNRows(pDst, pRes->info.rows, pSrc, startIdx, rows));

×

562

      probeIdx++;

×

566

  pRes->info.rows += rows;

×

568

  return TSDB_CODE_SUCCESS;

×

UNCOV

573

void hJoinAppendResToBlock(struct SOperatorInfo* pOperator, SSDataBlock* pRes, bool* allFetched) {

×

UNCOV

574

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

UNCOV

575

  SHJoinCtx* pCtx = &pJoin->ctx;

×

UNCOV

576

  SBufRowInfo* pStart = pCtx->pBuildRow;

×

UNCOV

577

  int32_t rowNum = 0;

×

UNCOV

578

  int32_t resNum = pRes->info.rows;

×

UNCOV

580

  while (pCtx->pBuildRow && (resNum < pRes->info.capacity)) {

×

UNCOV

581

    rowNum++;

×

UNCOV

582

    resNum++;

×

UNCOV

583

    pCtx->pBuildRow = pCtx->pBuildRow->next;

×

UNCOV

586

  pJoin->execInfo.resRows += rowNum;

×

UNCOV

588

  int32_t code = hJoinCopyResRowsToBlock(pJoin, rowNum, pStart, pRes);

×

UNCOV

589

  if (code) {

×

590

    pOperator->pTaskInfo->code = code;

×

591

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

×

UNCOV

594

  pRes->info.rows = resNum;

×

UNCOV

595

  *allFetched = pCtx->pBuildRow ? false : true;

×

UNCOV

596

×

UNCOV

599

bool hJoinCopyKeyColsDataToBuf(SHJoinTableCtx* pTable, int32_t rowIdx, size_t *pBufLen) {

×

UNCOV

600

  char *pData = NULL;

×

UNCOV

601

  size_t bufLen = 0;

×

UNCOV

603

  if (1 == pTable->keyNum) {

×

UNCOV

604

    if (colDataIsNull_s(pTable->keyCols[0].colData, rowIdx)) {

×

UNCOV

605

      return true;

×

UNCOV

607

    if (pTable->keyCols[0].vardata) {

×

UNCOV

608

      pData = pTable->keyCols[0].data + pTable->keyCols[0].offset[rowIdx];

×

UNCOV

609

      bufLen = varDataTLen(pData);

×

UNCOV

611

      pData = pTable->keyCols[0].data + pTable->keyCols[0].bytes * rowIdx;

×

UNCOV

612

      bufLen = pTable->keyCols[0].bytes;

×

UNCOV

614

    pTable->keyData = pData;

×

UNCOV

616

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

×

UNCOV

617

      if (colDataIsNull_s(pTable->keyCols[i].colData, rowIdx)) {

×

618

        return true;

×

UNCOV

620

      if (pTable->keyCols[i].vardata) {

×

UNCOV

621

        pData = pTable->keyCols[i].data + pTable->keyCols[i].offset[rowIdx];

×

UNCOV

622

        TAOS_MEMCPY(pTable->keyBuf + bufLen, pData, varDataTLen(pData));

×

UNCOV

623

        bufLen += varDataTLen(pData);

×

UNCOV

625

        pData = pTable->keyCols[i].data + pTable->keyCols[i].bytes * rowIdx;

×

UNCOV

626

        TAOS_MEMCPY(pTable->keyBuf + bufLen, pData, pTable->keyCols[i].bytes);

×

UNCOV

627

        bufLen += pTable->keyCols[i].bytes;

×

UNCOV

630

    pTable->keyData = pTable->keyBuf;

×

UNCOV

633

  if (pBufLen) {

×

UNCOV

634

    *pBufLen = bufLen;

×

UNCOV

637

  return false;

×

UNCOV

640

static int32_t hJoinSetKeyColsData(SSDataBlock* pBlock, SHJoinTableCtx* pTable) {

×

UNCOV

641

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

×

UNCOV

642

    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, pTable->keyCols[i].srcSlot);

×

UNCOV

643

    if (NULL == pCol) {

×

644

      qError("fail to get %d col, total:%d", pTable->keyCols[i].srcSlot, (int32_t)taosArrayGetSize(pBlock->pDataBlock));

×

645

      QRY_ERR_RET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

647

    if (pTable->keyCols[i].vardata != IS_VAR_DATA_TYPE(pCol->info.type))  {

×

648

      qError("column type mismatch, idx:%d, slotId:%d, type:%d, vardata:%d", i, pTable->keyCols[i].srcSlot, pCol->info.type, pTable->keyCols[i].vardata);

×

649

      return TSDB_CODE_INVALID_PARA;

×

UNCOV

651

    if (pTable->keyCols[i].bytes != pCol->info.bytes)  {

×

652

      qError("column bytes mismatch, idx:%d, slotId:%d, bytes:%d, %d", i, pTable->keyCols[i].srcSlot, pCol->info.bytes, pTable->keyCols[i].bytes);

×

653

      return TSDB_CODE_INVALID_PARA;

×

UNCOV

655

    pTable->keyCols[i].data = pCol->pData;

×

UNCOV

656

    if (pTable->keyCols[i].vardata) {

×

UNCOV

657

      pTable->keyCols[i].offset = pCol->varmeta.offset;

×

UNCOV

659

    pTable->keyCols[i].colData = pCol;

×

UNCOV

662

  return TSDB_CODE_SUCCESS;

×

UNCOV

665

static int32_t hJoinSetValColsData(SSDataBlock* pBlock, SHJoinTableCtx* pTable) {

×

UNCOV

666

  if (!pTable->valColExist) {

×

667

    return TSDB_CODE_SUCCESS;

×

UNCOV

669

  for (int32_t i = 0; i < pTable->valNum; ++i) {

×

UNCOV

670

    if (pTable->valCols[i].keyCol) {

×

671

      continue;

×

UNCOV

673

    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, pTable->valCols[i].srcSlot);

×

UNCOV

674

    if (NULL == pCol) {

×

675

      qError("fail to get %d col, total:%d", pTable->valCols[i].srcSlot, (int32_t)taosArrayGetSize(pBlock->pDataBlock));

×

676

      QRY_ERR_RET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

678

    if (pTable->valCols[i].vardata != IS_VAR_DATA_TYPE(pCol->info.type))  {

×

679

      qError("column type mismatch, idx:%d, slotId:%d, type:%d, vardata:%d", i, pTable->valCols[i].srcSlot, pCol->info.type, pTable->valCols[i].vardata);

×

680

      return TSDB_CODE_INVALID_PARA;

×

UNCOV

682

    if (pTable->valCols[i].bytes != pCol->info.bytes)  {

×

683

      qError("column bytes mismatch, idx:%d, slotId:%d, bytes:%d, %d", i, pTable->valCols[i].srcSlot, pCol->info.bytes, pTable->valCols[i].bytes);

×

684

      return TSDB_CODE_INVALID_PARA;

×

UNCOV

686

    if (!pTable->valCols[i].vardata) {

×

UNCOV

687

      pTable->valCols[i].bitMap = pCol->nullbitmap;

×

UNCOV

689

    pTable->valCols[i].data = pCol->pData;

×

UNCOV

690

    if (pTable->valCols[i].vardata) {

×

691

      pTable->valCols[i].offset = pCol->varmeta.offset;

×

UNCOV

695

  return TSDB_CODE_SUCCESS;

×

UNCOV

701

  if (!pTable->valColExist) {

×

702

    return;

×

UNCOV

705

  char *pData = NULL;

×

UNCOV

706

  size_t bufLen = pTable->valBitMapSize;

×

UNCOV

707

  TAOS_MEMSET(pTable->valData, 0, pTable->valBitMapSize);

×

UNCOV

708

  for (int32_t i = 0, m = 0; i < pTable->valNum; ++i) {

×

UNCOV

709

    if (pTable->valCols[i].keyCol) {

×

710

      continue;

×

UNCOV

712

    if (pTable->valCols[i].vardata) {

×

713

      if (-1 == pTable->valCols[i].offset[rowIdx]) {

×

714

        colDataSetNull_f(pTable->valData, m);

×

716

        pData = pTable->valCols[i].data + pTable->valCols[i].offset[rowIdx];

×

717

        TAOS_MEMCPY(pTable->valData + bufLen, pData, varDataTLen(pData));

×

718

        bufLen += varDataTLen(pData);

×

UNCOV

721

      if (colDataIsNull_f(pTable->valCols[i].bitMap, rowIdx)) {

×

722

        colDataSetNull_f(pTable->valData, m);

×

UNCOV

724

        pData = pTable->valCols[i].data + pTable->valCols[i].bytes * rowIdx;

×

UNCOV

725

        TAOS_MEMCPY(pTable->valData + bufLen, pData, pTable->valCols[i].bytes);

×

UNCOV

726

        bufLen += pTable->valCols[i].bytes;

×

UNCOV

729

    m++;

×

UNCOV

735

  if (0 == bufSize) {

×

736

    pRow->pageId = -1;

×

737

    return TSDB_CODE_SUCCESS;

×

UNCOV

740

  if (bufSize > HASH_JOIN_DEFAULT_PAGE_SIZE) {

×

741

    qError("invalid join value buf size:%d", bufSize);

×

742

    return TSDB_CODE_INVALID_PARA;

×

745

  do {

×

UNCOV

746

    SBufPageInfo* page = taosArrayGetLast(pPages);

×

UNCOV

747

    if ((page->pageSize - page->offset) >= bufSize) {

×

UNCOV

748

      *pBuf = page->data + page->offset;

×

UNCOV

749

      pRow->pageId = taosArrayGetSize(pPages) - 1;

×

UNCOV

750

      pRow->offset = page->offset;

×

UNCOV

751

      page->offset += bufSize;

×

UNCOV

752

      return TSDB_CODE_SUCCESS;

×

755

    int32_t code = hJoinAddPageToBufs(pPages);

×

756

    if (code) {

×

757

      return code;

×

UNCOV

763

  if (NULL == pTable->valVarCols) {

×

UNCOV

764

    return pTable->valBufSize;

×

767

  int32_t* varColIdx = NULL;

×

768

  int32_t bufLen = pTable->valBufSize;

×

769

  int32_t varColNum = taosArrayGetSize(pTable->valVarCols);

×

770

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

×

771

    varColIdx = taosArrayGet(pTable->valVarCols, i);

×

772

    char* pData = pTable->valCols[*varColIdx].data + pTable->valCols[*varColIdx].offset[rowIdx];

×

773

    bufLen += varDataTLen(pData);

×

776

  return bufLen;

×

UNCOV

780

static int32_t hJoinAddRowToHashImpl(SHJoinOperatorInfo* pJoin, SGroupData* pGroup, SHJoinTableCtx* pTable, size_t keyLen, int32_t rowIdx) {

×

UNCOV

781

  SGroupData group = {0};

×

UNCOV

782

  SBufRowInfo* pRow = NULL;

×

UNCOV

784

  if (NULL == pGroup) {

×

UNCOV

785

    group.rows = taosMemoryMalloc(sizeof(SBufRowInfo));

×

UNCOV

786

    if (NULL == group.rows) {

×

787

      return terrno;

×

UNCOV

789

    pRow = group.rows;

×

UNCOV

791

    pRow = taosMemoryMalloc(sizeof(SBufRowInfo));

×

UNCOV

792

    if (NULL == pRow) {

×

793

      return terrno;

×

UNCOV

797

  int32_t code = hJoinGetValBufFromPages(pJoin->pRowBufs, hJoinGetValBufSize(pTable, rowIdx), &pTable->valData, pRow);

×

UNCOV

798

  if (code) {

×

799

    taosMemoryFree(pRow);

×

800

    return code;

×

UNCOV

803

  if (NULL == pGroup) {

×

UNCOV

804

    pRow->next = NULL;

×

UNCOV

805

    if (tSimpleHashPut(pJoin->pKeyHash, pTable->keyData, keyLen, &group, sizeof(group))) {

×

806

      taosMemoryFree(pRow);

×

807

      return terrno;

×

UNCOV

810

    pRow->next = pGroup->rows;

×

UNCOV

811

    pGroup->rows = pRow;

×

UNCOV

814

  return TSDB_CODE_SUCCESS;

×

UNCOV

817

static int32_t hJoinAddRowToHash(SHJoinOperatorInfo* pJoin, SSDataBlock* pBlock, size_t keyLen, int32_t rowIdx) {

×

UNCOV

818

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

UNCOV

819

  int32_t code = hJoinSetValColsData(pBlock, pBuild);

×

UNCOV

820

  if (code) {

×

821

    return code;

×

UNCOV

824

  SGroupData* pGroup = tSimpleHashGet(pJoin->pKeyHash, pBuild->keyData, keyLen);

×

UNCOV

825

  code = hJoinAddRowToHashImpl(pJoin, pGroup, pBuild, keyLen, rowIdx);

×

UNCOV

826

  if (code) {

×

827

    return code;

×

UNCOV

832

  return TSDB_CODE_SUCCESS;

×

835

static bool hJoinFilterTimeRange(SSDataBlock* pBlock, STimeWindow* pRange, int32_t primSlot, int32_t* startIdx, int32_t* endIdx) {

×

836

  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, primSlot);

×

837

  if (NULL == pCol) {

×

838

    qError("hash join can't get prim col, slot:%d, slotNum:%d", primSlot, (int32_t)taosArrayGetSize(pBlock->pDataBlock));

×

839

    return false;

×

842

  TSKEY skey = *(TSKEY*)colDataGetData(pCol, 0);

×

843

  TSKEY ekey = *(TSKEY*)colDataGetData(pCol, (pBlock->info.rows - 1));

×

845

  if (ekey < pRange->skey || skey > pRange->ekey) {

×

846

    return false;

×

849

  if (skey >= pRange->skey && ekey <= pRange->ekey) {

×

850

    *startIdx = 0;

×

851

    *endIdx = pBlock->info.rows - 1;

×

852

    return true;

×

855

  if (skey < pRange->skey && ekey > pRange->ekey) {

×

856

    TSKEY *pStart = (TSKEY*)taosbsearch(&pRange->skey, pCol->pData, pBlock->info.rows, sizeof(TSKEY), compareInt64Val, TD_GE);

×

857

    TSKEY *pEnd = (TSKEY*)taosbsearch(&pRange->ekey, pCol->pData, pBlock->info.rows, sizeof(TSKEY), compareInt64Val, TD_LE);

×

858

    *startIdx = ((uint64_t)pStart - (uint64_t)pCol->pData) / sizeof(int64_t);

×

859

    *endIdx = ((uint64_t)pEnd - (uint64_t)pCol->pData) / sizeof(int64_t);

×

860

    return true;

×

863

  if (skey >= pRange->skey) {

×

864

    TSKEY *pEnd = (TSKEY*)taosbsearch(&pRange->ekey, pCol->pData, pBlock->info.rows, sizeof(TSKEY), compareInt64Val, TD_LE);

×

865

    *startIdx = 0;

×

866

    *endIdx = ((uint64_t)pEnd - (uint64_t)pCol->pData) / sizeof(int64_t);

×

867

    return true;

×

870

  TSKEY *pStart = (TSKEY*)taosbsearch(&pRange->skey, pCol->pData, pBlock->info.rows, sizeof(TSKEY), compareInt64Val, TD_GE);

×

871

  *startIdx = ((uint64_t)pStart - (uint64_t)pCol->pData) / sizeof(int64_t);

×

872

  *endIdx = pBlock->info.rows - 1;

×

874

  return true;

×

UNCOV

877

static int32_t hJoinAddBlockRowsToHash(SSDataBlock* pBlock, SHJoinOperatorInfo* pJoin) {

×

UNCOV

878

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

UNCOV

879

  int32_t startIdx = 0, endIdx = pBlock->info.rows - 1;

×

UNCOV

880

  if (pBuild->hasTimeRange && !hJoinFilterTimeRange(pBlock, &pJoin->tblTimeRange, pBuild->primCol->srcSlot, &startIdx, &endIdx)) {

×

881

    return TSDB_CODE_SUCCESS;

×

UNCOV

884

  HJ_ERR_RET(hJoinLaunchPrimExpr(pBlock, pBuild, startIdx, endIdx));

×

UNCOV

886

  int32_t code = hJoinSetKeyColsData(pBlock, pBuild);

×

UNCOV

887

  if (code) {

×

888

    return code;

×

UNCOV

891

  size_t bufLen = 0;

×

UNCOV

892

  for (int32_t i = startIdx; i <= endIdx; ++i) {

×

UNCOV

893

    if (hJoinCopyKeyColsDataToBuf(pBuild, i, &bufLen)) {

×

UNCOV

894

      continue;

×

UNCOV

896

    code = hJoinAddRowToHash(pJoin, pBlock, bufLen, i);

×

UNCOV

897

    if (code) {

×

898

      return code;

×

UNCOV

902

  return code;

×

UNCOV

905

static int32_t hJoinBuildHash(struct SOperatorInfo* pOperator, bool* queryDone) {

×

UNCOV

906

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

UNCOV

907

  SSDataBlock*        pBlock = NULL;

×

UNCOV

908

  int32_t             code = TSDB_CODE_SUCCESS;

×

UNCOV

911

    pBlock = getNextBlockFromDownstream(pOperator, pJoin->pBuild->downStreamIdx);

×

UNCOV

912

    if (NULL == pBlock) {

×

UNCOV

913

      break;

×

UNCOV

916

    pJoin->execInfo.buildBlkNum++;

×

UNCOV

917

    pJoin->execInfo.buildBlkRows += pBlock->info.rows;

×

UNCOV

919

    code = hJoinAddBlockRowsToHash(pBlock, pJoin);

×

UNCOV

920

    if (code) {

×

921

      return code;

×

UNCOV

925

  if (IS_INNER_NONE_JOIN(pJoin->joinType, pJoin->subType) && tSimpleHashGetSize(pJoin->pKeyHash) <= 0) {

×

UNCOV

926

    hJoinSetDone(pOperator);

×

UNCOV

927

    *queryDone = true;

×

UNCOV

932

  return TSDB_CODE_SUCCESS;

×

UNCOV

935

static int32_t hJoinPrepareStart(struct SOperatorInfo* pOperator, SSDataBlock* pBlock) {

×

UNCOV

936

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

UNCOV

937

  SHJoinTableCtx* pProbe = pJoin->pProbe;

×

UNCOV

938

  int32_t startIdx = 0, endIdx = pBlock->info.rows - 1;

×

UNCOV

939

  if (pProbe->hasTimeRange && !hJoinFilterTimeRange(pBlock, &pJoin->tblTimeRange, pProbe->primCol->srcSlot, &startIdx, &endIdx)) {

×

940

    if (!IS_INNER_NONE_JOIN(pJoin->joinType, pJoin->subType)) {

×

941

      pJoin->ctx.probeEndIdx = -1;

×

942

      pJoin->ctx.probePostIdx = 0;

×

943

      pJoin->ctx.pProbeData = pBlock;

×

944

      pJoin->ctx.rowRemains = true;

×

945

      pJoin->ctx.probePhase = E_JOIN_PHASE_POST;

×

947

      HJ_ERR_RET((*pJoin->joinFp)(pOperator));

×

950

    return TSDB_CODE_SUCCESS;

×

UNCOV

953

  HJ_ERR_RET(hJoinLaunchPrimExpr(pBlock, pProbe, startIdx, endIdx));

×

UNCOV

955

  int32_t code = hJoinSetKeyColsData(pBlock, pProbe);

×

UNCOV

956

  if (code) {

×

957

    return code;

×

UNCOV

959

  code = hJoinSetValColsData(pBlock, pProbe);

×

UNCOV

960

  if (code) {

×

961

    return code;

×

UNCOV

964

  pJoin->ctx.probeStartIdx = startIdx;

×

UNCOV

965

  pJoin->ctx.probeEndIdx = endIdx;

×

UNCOV

966

  pJoin->ctx.pBuildRow = NULL;

×

UNCOV

967

  pJoin->ctx.pProbeData = pBlock;

×

UNCOV

968

  pJoin->ctx.rowRemains = true;

×

UNCOV

969

  pJoin->ctx.probePreIdx = 0;

×

UNCOV

970

  pJoin->ctx.probePostIdx = endIdx + 1;

×

UNCOV

972

  if (!IS_INNER_NONE_JOIN(pJoin->joinType, pJoin->subType) && startIdx > 0) {

×

973

    pJoin->ctx.probePhase = E_JOIN_PHASE_PRE;

×

UNCOV

975

    pJoin->ctx.probePhase = E_JOIN_PHASE_CUR;

×

UNCOV

978

  HJ_ERR_RET((*pJoin->joinFp)(pOperator));

×

UNCOV

980

  return TSDB_CODE_SUCCESS;

×

UNCOV

983

void hJoinSetDone(struct SOperatorInfo* pOperator) {

×

UNCOV

984

  setOperatorCompleted(pOperator);

×

UNCOV

986

  SHJoinOperatorInfo* pInfo = pOperator->info;

×

UNCOV

987

  hJoinDestroyKeyHash(&pInfo->pKeyHash);

×

UNCOV

989

  qDebug("hash Join done");

×

UNCOV

990

×

UNCOV

992

static int32_t hJoinMainProcess(struct SOperatorInfo* pOperator, SSDataBlock** pResBlock) {

×

UNCOV

993

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

UNCOV

994

  SExecTaskInfo*      pTaskInfo = pOperator->pTaskInfo;

×

UNCOV

995

  int32_t             code = TSDB_CODE_SUCCESS;

×

UNCOV

996

  int32_t             lino = 0;

×

UNCOV

997

  SSDataBlock*        pRes = pJoin->finBlk;

×

UNCOV

998

  int64_t             st = 0;

×

UNCOV

1000

  QRY_PARAM_CHECK(pResBlock);

×

UNCOV

1001

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

×

UNCOV

1002

    st = taosGetTimestampUs();

×

UNCOV

1005

  if (pOperator->status == OP_EXEC_DONE) {

×

UNCOV

1006

    pRes->info.rows = 0;

×

UNCOV

1007

    goto _end;

×

UNCOV

1010

  if (!pJoin->keyHashBuilt) {

×

UNCOV

1011

    pJoin->keyHashBuilt = true;

×

UNCOV

1013

    bool queryDone = false;

×

UNCOV

1014

    code = hJoinBuildHash(pOperator, &queryDone);

×

UNCOV

1015

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1017

    if (queryDone) {

×

UNCOV

1018

      goto _end;

×

UNCOV

1022

  blockDataCleanup(pRes);

×

UNCOV

1024

  if (pJoin->ctx.rowRemains) {

×

1025

    code = (*pJoin->joinFp)(pOperator);

×

1026

    QUERY_CHECK_CODE(code, lino, _end);

×

1028

    if (pRes->info.rows > 0 && pJoin->pFinFilter != NULL) {

×

1029

      code = doFilter(pRes, pJoin->pFinFilter, NULL);

×

1030

      QUERY_CHECK_CODE(code, lino, _end);

×

1033

    if (pRes->info.rows > 0) {

×

1034

      *pResBlock = pRes;

×

1035

      return code;

×

UNCOV

1039

  while (true) {

×

UNCOV

1040

    SSDataBlock* pBlock = getNextBlockFromDownstream(pOperator, pJoin->pProbe->downStreamIdx);

×

UNCOV

1041

    if (NULL == pBlock) {

×

UNCOV

1042

      hJoinSetDone(pOperator);

×

UNCOV

1043

      break;

×

UNCOV

1046

    pJoin->execInfo.probeBlkNum++;

×

UNCOV

1047

    pJoin->execInfo.probeBlkRows += pBlock->info.rows;

×

UNCOV

1049

    code = hJoinPrepareStart(pOperator, pBlock);

×

UNCOV

1050

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1052

    if (!hJoinBlkReachThreshold(pJoin, pRes->info.rows)) {

×

UNCOV

1053

      continue;

×

1056

    if (pRes->info.rows > 0 && pJoin->pFinFilter != NULL) {

×

1057

      code = doFilter(pRes, pJoin->pFinFilter, NULL);

×

1058

      QUERY_CHECK_CODE(code, lino, _end);

×

1061

    if (pRes->info.rows > 0) {

×

1062

      break;

×

UNCOV

1066

_end:

×

UNCOV

1067

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

×

UNCOV

1068

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

×

UNCOV

1070

  if (code != TSDB_CODE_SUCCESS) {

×

1071

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

×

1072

    pTaskInfo->code = code;

×

1073

    T_LONG_JMP(pTaskInfo->env, code);

×

UNCOV

1075

  if (pRes->info.rows > 0) {

×

UNCOV

1076

    *pResBlock = pRes;

×

UNCOV

1079

  return code;

×

UNCOV

1082

static void destroyHashJoinOperator(void* param) {

×

UNCOV

1083

  SHJoinOperatorInfo* pJoinOperator = (SHJoinOperatorInfo*)param;

×

UNCOV

1084

  qDebug("hashJoin exec info, buildBlk:%" PRId64 ", buildRows:%" PRId64 ", probeBlk:%" PRId64 ", probeRows:%" PRId64 ", resRows:%" PRId64,

×

UNCOV

1088

  hJoinDestroyKeyHash(&pJoinOperator->pKeyHash);

×

UNCOV

1090

  hJoinFreeTableInfo(&pJoinOperator->tbs[0]);

×

UNCOV

1091

  hJoinFreeTableInfo(&pJoinOperator->tbs[1]);

×

UNCOV

1092

  blockDataDestroy(pJoinOperator->finBlk);

×

UNCOV

1093

  pJoinOperator->finBlk = NULL;

×

UNCOV

1094

  taosMemoryFreeClear(pJoinOperator->pResColMap);

×

UNCOV

1095

  taosArrayDestroyEx(pJoinOperator->pRowBufs, hJoinFreeBufPage);

×

UNCOV

1097

  taosMemoryFreeClear(param);

×

UNCOV

1098

×

UNCOV

1100

int32_t hJoinHandleConds(SHJoinOperatorInfo* pJoin, SHashJoinPhysiNode* pJoinNode) {

×

UNCOV

1101

  switch (pJoin->joinType) {

×

UNCOV

1102

    case JOIN_TYPE_INNER: {

×

UNCOV

1103

      SNode* pCond = NULL;

×

UNCOV

1104

      if (pJoinNode->pFullOnCond != NULL) {

×

1105

        if (pJoinNode->node.pConditions != NULL) {

×

1106

          HJ_ERR_RET(mergeJoinConds(&pJoinNode->pFullOnCond, &pJoinNode->node.pConditions));

×

1108

        pCond = pJoinNode->pFullOnCond;

×

UNCOV

1109

      } else if (pJoinNode->node.pConditions != NULL) {

×

1110

        pCond = pJoinNode->node.pConditions;

×

UNCOV

1113

      HJ_ERR_RET(filterInitFromNode(pCond, &pJoin->pFinFilter, 0));

×

UNCOV

1114

      break;

×

1116

    case JOIN_TYPE_LEFT:

×

1119

      if (pJoinNode->pFullOnCond != NULL) {

×

1120

        HJ_ERR_RET(filterInitFromNode(pJoinNode->pFullOnCond, &pJoin->pPreFilter, 0));

×

1122

      if (pJoinNode->node.pConditions != NULL) {

×

1123

        HJ_ERR_RET(filterInitFromNode(pJoinNode->node.pConditions, &pJoin->pFinFilter, 0));

×

1125

      break;

×

1126

    default:

×

1127

      break;

×

UNCOV

1130

  return TSDB_CODE_SUCCESS;

×

UNCOV

1133

static uint32_t hJoinGetFinBlkCapacity(SHJoinOperatorInfo* pJoin, SHashJoinPhysiNode* pJoinNode) {

×

UNCOV

1134

  uint32_t maxRows = TMAX(HJOIN_DEFAULT_BLK_ROWS_NUM, HJOIN_BLK_SIZE_LIMIT/pJoinNode->node.pOutputDataBlockDesc->totalRowSize);

×

UNCOV

1135

  if (INT64_MAX != pJoin->ctx.limit && NULL == pJoin->pFinFilter) {

×

1136

    uint32_t limitMaxRows = pJoin->ctx.limit / HJOIN_BLK_THRESHOLD_RATIO + 1;

×

1137

    return (maxRows > limitMaxRows) ? limitMaxRows : maxRows;

×

UNCOV

1140

  return maxRows;

×

UNCOV

1144

int32_t hJoinInitResBlocks(SHJoinOperatorInfo* pJoin, SHashJoinPhysiNode* pJoinNode) {

×

UNCOV

1145

  pJoin->finBlk = createDataBlockFromDescNode(pJoinNode->node.pOutputDataBlockDesc);

×

UNCOV

1146

  if (NULL == pJoin->finBlk) {

×

1147

    QRY_ERR_RET(terrno);

×

UNCOV

1150

  int32_t code = blockDataEnsureCapacity(pJoin->finBlk, hJoinGetFinBlkCapacity(pJoin, pJoinNode));

×

UNCOV

1151

  if (TSDB_CODE_SUCCESS != code) {

×

1152

    QRY_ERR_RET(code);

×

UNCOV

1155

  if (NULL != pJoin->pPreFilter) {

×

1156

    pJoin->midBlk = NULL;

×

1157

    code = createOneDataBlock(pJoin->finBlk, false, &pJoin->midBlk);

×

1158

    if (code) {

×

1159

      QRY_ERR_RET(code);

×

1162

    code = blockDataEnsureCapacity(pJoin->midBlk, pJoin->finBlk->info.capacity);

×

1163

    if (TSDB_CODE_SUCCESS != code) {

×

1164

      QRY_ERR_RET(code);

×

UNCOV

1168

  pJoin->blkThreshold = pJoin->finBlk->info.capacity * HJOIN_BLK_THRESHOLD_RATIO;

×

UNCOV

1169

  return TSDB_CODE_SUCCESS;

×

UNCOV

1173

int32_t createHashJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream,

×

UNCOV

1175

  QRY_PARAM_CHECK(pOptrInfo);

×

UNCOV

1177

  int32_t             code = TSDB_CODE_SUCCESS;

×

UNCOV

1178

  SHJoinOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SHJoinOperatorInfo));

×

UNCOV

1179

  SOperatorInfo*      pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));

×

UNCOV

1180

  if (pOperator == NULL || pInfo == NULL) {

×

1181

    code = terrno;

×

1182

    goto _return;

×

UNCOV

1185

  pInfo->tblTimeRange.skey = pJoinNode->timeRange.skey;

×

UNCOV

1186

  pInfo->tblTimeRange.ekey = pJoinNode->timeRange.ekey;

×

UNCOV

1188

  pInfo->ctx.limit = (pJoinNode->node.pLimit && ((SLimitNode*)pJoinNode->node.pLimit)->limit) ? ((SLimitNode*)pJoinNode->node.pLimit)->limit->datum.i : INT64_MAX;

×

UNCOV

1190

  setOperatorInfo(pOperator, "HashJoinOperator", QUERY_NODE_PHYSICAL_PLAN_HASH_JOIN, false, OP_NOT_OPENED, pInfo, pTaskInfo);

×

UNCOV

1192

  HJ_ERR_JRET(hJoinInitTableInfo(pInfo, pJoinNode, pDownstream, 0, &pJoinNode->inputStat[0]));

×

UNCOV

1193

  HJ_ERR_JRET(hJoinInitTableInfo(pInfo, pJoinNode, pDownstream, 1, &pJoinNode->inputStat[1]));

×

UNCOV

1195

  hJoinSetBuildAndProbeTable(pInfo, pJoinNode);

×

UNCOV

1197

  HJ_ERR_JRET(hJoinBuildResColsMap(pInfo, pJoinNode));

×

UNCOV

1199

  HJ_ERR_JRET(hJoinInitBufPages(pInfo));

×

UNCOV

1201

  size_t hashCap = pInfo->pBuild->inputStat.inputRowNum > 0 ? (pInfo->pBuild->inputStat.inputRowNum * 1.5) : 1024;

×

UNCOV

1202

  pInfo->pKeyHash = tSimpleHashInit(hashCap, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));

×

UNCOV

1203

  if (pInfo->pKeyHash == NULL) {

×

1204

    code = terrno;

×

1205

    goto _return;

×

UNCOV

1208

  HJ_ERR_JRET(hJoinHandleConds(pInfo, pJoinNode));

×

UNCOV

1210

  HJ_ERR_JRET(hJoinInitResBlocks(pInfo, pJoinNode));

×

UNCOV

1212

  HJ_ERR_JRET(hJoinSetImplFp(pInfo));

×

UNCOV

1214

  HJ_ERR_JRET(appendDownstream(pOperator, pDownstream, numOfDownstream));

×

UNCOV

1216

  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, hJoinMainProcess, NULL, destroyHashJoinOperator, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);

×

UNCOV

1218

  qDebug("create hash Join operator done");

×

UNCOV

1220

  *pOptrInfo = pOperator;

×

UNCOV

1221

  return TSDB_CODE_SUCCESS;

×

1223

_return:

×

1225

  if (pInfo != NULL) {

×

1226

    destroyHashJoinOperator(pInfo);

×

1228

  destroyOperatorAndDownstreams(pOperator, pDownstream, numOfDownstream);

×

1229

  pTaskInfo->code = code;

×

1230

  return code;

×

taosdata / TDengine / #3621

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

taosdata / TDengine / #3621

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

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