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

32

bool hJoinBlkReachThreshold(SHJoinOperatorInfo* pInfo, int64_t blkRows) {

×

33

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

×

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;

×

83

int32_t hJoinSetImplFp(SHJoinOperatorInfo* pJoin) {

×

84

  switch (pJoin->joinType) {

×

85

    case JOIN_TYPE_INNER:

×

86

      pJoin->joinFp = hInnerJoinDo;

×

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;

×

103

  return TSDB_CODE_SUCCESS;

×

107

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

×

108

  if (NULL == pTable->primExpr) {

×

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;

×

153

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

×

154

  pTable->keyNum = LIST_LENGTH(pList);

×

156

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

×

157

  if (NULL == pTable->keyCols) {

×

158

    return terrno;

×

161

  int64_t bufSize = 0;

×

162

  int32_t i = 0;

×

163

  SNode* pNode = NULL;

×

164

  FOREACH(pNode, pList) {

×

165

    SColumnNode* pColNode = (SColumnNode*)pNode;

×

166

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

×

167

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

×

168

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

×

169

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

×

170

    ++i;

×

173

  if (pTable->keyNum > 1) {

×

174

    pTable->keyBuf = taosMemoryMalloc(bufSize);

×

175

    if (NULL == pTable->keyBuf) {

×

176

      return terrno;

×

180

  return TSDB_CODE_SUCCESS;

×

183

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

×

184

  *colNum = 0;

×

186

  SNode* pNode = NULL;

×

187

  FOREACH(pNode, pList) {

×

188

    STargetNode* pTarget = (STargetNode*)pNode;

×

189

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

×

190

    if (pCol->dataBlockId == blkId) {

×

191

      (*colNum)++;

×

194

×

196

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

×

197

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

×

198

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

×

199

      *pKeyIdx = i;

×

200

      return true;

×

204

  *pKeyIdx = -1;

×

205

  return false;

×

208

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

×

209

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

×

210

  if (pTable->valNum == 0) {

×

211

    return TSDB_CODE_SUCCESS;

×

214

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

×

215

  if (NULL == pTable->valCols) {

×

216

    return terrno;

×

219

  int32_t i = 0;

×

220

  int32_t colNum = 0;

×

221

  SNode* pNode = NULL;

×

222

  FOREACH(pNode, pList) {

×

223

    STargetNode* pTarget = (STargetNode*)pNode;

×

224

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

×

225

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

×

226

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

×

227

        pTable->valColExist = true;

×

228

        colNum++;

×

230

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

×

231

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

×

232

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

×

233

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

×

234

        if (NULL == pTable->valVarCols) {

×

235

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

×

236

          if (NULL == pTable->valVarCols) {

×

237

            return terrno;

×

240

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

×

241

          return terrno;

×

244

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

×

245

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

×

246

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

×

248

      i++;

×

252

  pTable->valBitMapSize = BitmapLen(colNum);

×

253

  pTable->valBufSize += pTable->valBitMapSize;

×

255

  return TSDB_CODE_SUCCESS;

×

258

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

×

259

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

×

260

  if (NULL == pTable->primCol) {

×

261

    return terrno;

×

264

  pTable->primCol->srcSlot = slotId;

×

266

  return TSDB_CODE_SUCCESS;

×

270

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

×

271

  if (NULL == pNode) {

×

272

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

×

273

    return TSDB_CODE_SUCCESS;

×

276

  if (QUERY_NODE_TARGET != nodeType(pNode)) {

×

277

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

280

  STargetNode* pTarget = (STargetNode*)pNode;

×

281

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

×

282

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

285

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

×

286

  if (FUNCTION_TYPE_TIMETRUNCATE != pFunc->funcType) {

×

287

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

290

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

×

291

    return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

294

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

×

295

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

×

296

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

×

298

  pCtx->truncateUnit = pUnit->typeData;

×

299

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

×

300

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

×

303

  pCtx->targetSlotId = pTarget->slotId;

×

305

  return TSDB_CODE_SUCCESS;

×

309

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

×

310

  SNodeList* pKeyList = NULL;

×

311

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

×

312

  pTable->downStream = pDownstream[idx];

×

313

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

×

314

  if (0 == idx) {

×

315

    pKeyList = pJoinNode->pOnLeft;

×

316

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

×

318

    pKeyList = pJoinNode->pOnRight;

×

319

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

×

322

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

×

324

  int32_t code = hJoinInitKeyColsInfo(pTable, pKeyList);

×

325

  if (code) {

×

326

    return code;

×

328

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

×

329

  if (code) {

×

330

    return code;

×

333

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

×

335

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

×

337

  return TSDB_CODE_SUCCESS;

×

340

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

×

341

  int32_t buildIdx = 0;

×

342

  int32_t probeIdx = 1;

×

344

  pInfo->joinType = pJoinNode->joinType;

×

345

  pInfo->subType = pJoinNode->subType;

×

347

  switch (pInfo->joinType) {

×

348

    case JOIN_TYPE_INNER:

×

350

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

×

351

        buildIdx = 0;

×

352

        probeIdx = 1;

×

354

        buildIdx = 1;

×

355

        probeIdx = 0;

×

357

      break;

×

358

    case JOIN_TYPE_LEFT:

×

359

      buildIdx = 1;

×

360

      probeIdx = 0;

×

361

      break;

×

362

    case JOIN_TYPE_RIGHT:

×

363

      buildIdx = 0;

×

364

      probeIdx = 1;

×

365

      break;

×

366

    default:

×

367

      break;

×

370

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

×

371

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

×

373

  pInfo->pBuild->downStreamIdx = buildIdx;

×

374

  pInfo->pProbe->downStreamIdx = probeIdx;

×

376

  if (0 == buildIdx) {

×

377

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

×

378

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

×

380

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

×

381

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

×

383

×

385

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

×

386

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

×

387

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

×

388

  if (NULL == pInfo->pResColMap) {

×

389

    return terrno;

×

392

  SNode* pNode = NULL;

×

393

  int32_t i = 0;

×

394

  FOREACH(pNode, pJoinNode->pTargets) {

×

395

    STargetNode* pTarget = (STargetNode*)pNode;

×

396

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

×

397

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

×

398

      pInfo->pResColMap[i] = 1;

×

401

    i++;

×

404

  return TSDB_CODE_SUCCESS;

×

410

  page.pageSize = HASH_JOIN_DEFAULT_PAGE_SIZE;

×

411

  page.offset = 0;

×

412

  page.data = taosMemoryMalloc(page.pageSize);

×

413

  if (NULL == page.data) {

×

414

    return terrno;

×

417

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

×

418

    return terrno;

×

420

  return TSDB_CODE_SUCCESS;

×

423

static int32_t hJoinInitBufPages(SHJoinOperatorInfo* pInfo) {

×

424

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

×

425

  if (NULL == pInfo->pRowBufs) {

×

426

    return terrno;

×

429

  return hJoinAddPageToBufs(pInfo->pRowBufs);

×

432

static void hJoinFreeTableInfo(SHJoinTableCtx* pTable) {

×

433

  taosMemoryFreeClear(pTable->keyCols);

×

434

  taosMemoryFreeClear(pTable->keyBuf);

×

435

  taosMemoryFreeClear(pTable->valCols);

×

436

  taosArrayDestroy(pTable->valVarCols);

×

437

  taosMemoryFree(pTable->primCol);

×

438

×

440

static void hJoinFreeBufPage(void* param) {

×

441

  SBufPageInfo* pInfo = (SBufPageInfo*)param;

×

442

  taosMemoryFree(pInfo->data);

×

443

×

445

static void hJoinDestroyKeyHash(SSHashObj** ppHash) {

×

446

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

×

447

    return;

×

450

  void*   pIte = NULL;

×

451

  int32_t iter = 0;

×

452

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

×

453

    SGroupData* pGroup = pIte;

×

454

    SBufRowInfo* pRow = pGroup->rows;

×

455

    SBufRowInfo* pNext = NULL;

×

456

    while (pRow) {

×

457

      pNext = pRow->next;

×

458

      taosMemoryFree(pRow);

×

459

      pRow = pNext;

×

463

  tSimpleHashCleanup(*ppHash);

×

464

  *ppHash = NULL;

×

468

  *ppData = NULL;

×

470

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

×

471

    return TSDB_CODE_SUCCESS;

×

473

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

×

474

  if (NULL == pPage) {

×

475

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

×

476

    QRY_ERR_RET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

479

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

×

481

  return TSDB_CODE_SUCCESS;

×

484

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

×

485

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

486

  SHJoinTableCtx* pProbe = pJoin->pProbe;

×

487

  int32_t buildIdx = 0, buildValIdx = 0;

×

488

  int32_t probeIdx = 0;

×

489

  SBufRowInfo* pRow = pStart;

×

490

  int32_t code = 0;

×

491

  char* pData = NULL;

×

493

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

×

494

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

×

496

    char* pValData = pData + pBuild->valBitMapSize;

×

497

    char* pKeyData = pProbe->keyData;

×

498

    buildIdx = buildValIdx = probeIdx = 0;

×

499

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

×

500

      if (pJoin->pResColMap[i]) {

×

501

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

×

502

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

×

503

          int32_t bufOffset = pProbe->keyCols[pBuild->valCols[buildIdx].keyColIdx].bufOffset;

×

504

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

×

505

          if (code) {

×

506

            return code;

×

509

          if (BMIsNull(pData, buildValIdx)) {

×

510

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

×

511

            if (code) {

×

512

              return code;

×

515

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

×

516

            if (code) {

×

517

              return code;

×

519

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

×

521

          buildValIdx++;

×

523

        buildIdx++;

×

524

      } else if (0 == r) {

×

525

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

×

526

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

×

528

        if (colDataIsNull_s(pSrc, pJoin->ctx.probeStartIdx)) {

×

529

          code = colDataCopyNItems(pDst, pRes->info.rows, NULL, rowNum, true);

×

531

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

×

533

        if (code) {

×

534

          return code;

×

536

        probeIdx++;

×

539

    pRow = pRow->next;

×

542

  return TSDB_CODE_SUCCESS;

×

545

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

×

546

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

547

  SHJoinTableCtx* pProbe = pJoin->pProbe;

×

548

  int32_t buildIdx = 0;

×

549

  int32_t probeIdx = 0;

×

550

  int32_t code = 0;

×

552

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

×

553

    if (pJoin->pResColMap[i]) {

×

554

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

×

555

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

×

557

      buildIdx++;

×

559

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

×

560

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

×

562

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

×

564

      probeIdx++;

×

568

  pRes->info.rows += rows;

×

570

  return TSDB_CODE_SUCCESS;

×

575

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

×

576

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

577

  SHJoinCtx* pCtx = &pJoin->ctx;

×

578

  SBufRowInfo* pStart = pCtx->pBuildRow;

×

579

  int32_t rowNum = 0;

×

580

  int32_t resNum = pRes->info.rows;

×

582

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

×

583

    rowNum++;

×

584

    resNum++;

×

585

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

×

588

  pJoin->execInfo.resRows += rowNum;

×

590

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

×

591

  if (code) {

×

592

    pOperator->pTaskInfo->code = code;

×

593

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

×

596

  pRes->info.rows = resNum;

×

597

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

×

598

×

601

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

×

602

  char *pData = NULL;

×

603

  size_t bufLen = 0;

×

605

  if (1 == pTable->keyNum) {

×

606

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

×

607

      return true;

×

609

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

×

610

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

×

611

      bufLen = varDataTLen(pData);

×

613

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

×

614

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

×

616

    pTable->keyData = pData;

×

618

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

×

619

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

×

620

        return true;

×

623

      pTable->keyCols[i].bufOffset = bufLen;

×

625

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

×

626

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

×

627

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

×

628

        bufLen += varDataTLen(pData);

×

630

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

×

631

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

×

632

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

×

635

    pTable->keyData = pTable->keyBuf;

×

638

  if (pBufLen) {

×

639

    *pBufLen = bufLen;

×

642

  return false;

×

645

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

×

646

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

×

647

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

×

648

    if (NULL == pCol) {

×

649

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

×

650

      QRY_ERR_RET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

652

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

×

653

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

×

654

      return TSDB_CODE_INVALID_PARA;

×

656

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

×

657

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

×

658

      return TSDB_CODE_INVALID_PARA;

×

660

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

×

661

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

×

662

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

×

664

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

×

667

  return TSDB_CODE_SUCCESS;

×

670

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

×

671

  if (!pTable->valColExist) {

×

672

    return TSDB_CODE_SUCCESS;

×

674

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

×

675

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

×

676

      continue;

×

678

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

×

679

    if (NULL == pCol) {

×

680

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

×

681

      QRY_ERR_RET(TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

683

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

×

684

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

×

685

      return TSDB_CODE_INVALID_PARA;

×

687

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

×

688

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

×

689

      return TSDB_CODE_INVALID_PARA;

×

691

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

×

692

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

×

694

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

×

695

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

×

696

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

×

700

  return TSDB_CODE_SUCCESS;

×

706

  if (!pTable->valColExist) {

×

707

    return;

×

710

  char *pData = NULL;

×

711

  size_t bufLen = pTable->valBitMapSize;

×

712

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

×

713

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

×

714

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

×

715

      continue;

×

717

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

×

718

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

×

719

        colDataSetNull_f(pTable->valData, m);

×

721

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

×

722

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

×

723

        bufLen += varDataTLen(pData);

×

726

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

×

727

        colDataSetNull_f(pTable->valData, m);

×

729

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

×

730

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

×

731

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

×

734

    m++;

×

740

  if (0 == bufSize) {

×

741

    pRow->pageId = -1;

×

742

    return TSDB_CODE_SUCCESS;

×

745

  if (bufSize > HASH_JOIN_DEFAULT_PAGE_SIZE) {

×

746

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

×

747

    return TSDB_CODE_INVALID_PARA;

×

750

  do {

×

751

    SBufPageInfo* page = taosArrayGetLast(pPages);

×

752

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

×

753

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

×

754

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

×

755

      pRow->offset = page->offset;

×

756

      page->offset += bufSize;

×

757

      return TSDB_CODE_SUCCESS;

×

760

    int32_t code = hJoinAddPageToBufs(pPages);

×

761

    if (code) {

×

762

      return code;

×

768

  if (NULL == pTable->valVarCols) {

×

769

    return pTable->valBufSize;

×

772

  int32_t* varColIdx = NULL;

×

773

  int32_t bufLen = pTable->valBufSize;

×

774

  int32_t varColNum = taosArrayGetSize(pTable->valVarCols);

×

775

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

×

776

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

×

777

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

×

778

      continue;

×

780

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

×

781

    bufLen += varDataTLen(pData);

×

784

  return bufLen;

×

788

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

×

789

  SGroupData group = {0};

×

790

  SBufRowInfo* pRow = NULL;

×

792

  if (NULL == pGroup) {

×

793

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

×

794

    if (NULL == group.rows) {

×

795

      return terrno;

×

797

    pRow = group.rows;

×

799

    pRow = taosMemoryMalloc(sizeof(SBufRowInfo));

×

800

    if (NULL == pRow) {

×

801

      return terrno;

×

805

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

×

806

  if (code) {

×

807

    taosMemoryFree(pRow);

×

808

    return code;

×

811

  if (NULL == pGroup) {

×

812

    pRow->next = NULL;

×

813

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

×

814

      taosMemoryFree(pRow);

×

815

      return terrno;

×

818

    pRow->next = pGroup->rows;

×

819

    pGroup->rows = pRow;

×

822

  return TSDB_CODE_SUCCESS;

×

825

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

×

826

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

827

  int32_t code = hJoinSetValColsData(pBlock, pBuild);

×

828

  if (code) {

×

829

    return code;

×

832

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

×

833

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

×

834

  if (code) {

×

835

    return code;

×

840

  return TSDB_CODE_SUCCESS;

×

843

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

×

844

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

×

845

  if (NULL == pCol) {

×

846

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

×

847

    return false;

×

850

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

×

851

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

×

853

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

×

854

    return false;

×

857

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

×

858

    *startIdx = 0;

×

859

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

×

860

    return true;

×

863

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

×

864

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

×

865

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

×

866

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

×

867

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

×

868

    return true;

×

871

  if (skey >= pRange->skey) {

×

872

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

×

873

    *startIdx = 0;

×

874

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

×

875

    return true;

×

878

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

×

879

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

×

880

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

×

882

  return true;

×

885

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

×

886

  SHJoinTableCtx* pBuild = pJoin->pBuild;

×

887

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

×

888

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

×

889

    return TSDB_CODE_SUCCESS;

×

892

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

×

894

  int32_t code = hJoinSetKeyColsData(pBlock, pBuild);

×

895

  if (code) {

×

896

    return code;

×

899

  size_t bufLen = 0;

×

900

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

×

901

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

×

902

      continue;

×

904

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

×

905

    if (code) {

×

906

      return code;

×

910

  return code;

×

913

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

×

914

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

915

  SSDataBlock*        pBlock = NULL;

×

916

  int32_t             code = TSDB_CODE_SUCCESS;

×

919

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

×

920

    if (NULL == pBlock) {

×

921

      break;

×

924

    pJoin->execInfo.buildBlkNum++;

×

925

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

×

927

    code = hJoinAddBlockRowsToHash(pBlock, pJoin);

×

928

    if (code) {

×

929

      return code;

×

933

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

×

934

    hJoinSetDone(pOperator);

×

935

    *queryDone = true;

×

940

  return TSDB_CODE_SUCCESS;

×

943

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

×

944

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

945

  SHJoinTableCtx* pProbe = pJoin->pProbe;

×

946

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

×

947

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

×

948

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

×

949

      pJoin->ctx.probeEndIdx = -1;

×

950

      pJoin->ctx.probePostIdx = 0;

×

951

      pJoin->ctx.pProbeData = pBlock;

×

952

      pJoin->ctx.rowRemains = true;

×

953

      pJoin->ctx.probePhase = E_JOIN_PHASE_POST;

×

955

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

×

958

    return TSDB_CODE_SUCCESS;

×

961

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

×

963

  int32_t code = hJoinSetKeyColsData(pBlock, pProbe);

×

964

  if (code) {

×

965

    return code;

×

967

  code = hJoinSetValColsData(pBlock, pProbe);

×

968

  if (code) {

×

969

    return code;

×

972

  pJoin->ctx.probeStartIdx = startIdx;

×

973

  pJoin->ctx.probeEndIdx = endIdx;

×

974

  pJoin->ctx.pBuildRow = NULL;

×

975

  pJoin->ctx.pProbeData = pBlock;

×

976

  pJoin->ctx.rowRemains = true;

×

977

  pJoin->ctx.probePreIdx = 0;

×

978

  pJoin->ctx.probePostIdx = endIdx + 1;

×

980

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

×

981

    pJoin->ctx.probePhase = E_JOIN_PHASE_PRE;

×

983

    pJoin->ctx.probePhase = E_JOIN_PHASE_CUR;

×

986

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

×

988

  return TSDB_CODE_SUCCESS;

×

991

void hJoinSetDone(struct SOperatorInfo* pOperator) {

×

992

  setOperatorCompleted(pOperator);

×

994

  SHJoinOperatorInfo* pInfo = pOperator->info;

×

995

  hJoinDestroyKeyHash(&pInfo->pKeyHash);

×

997

  qDebug("hash Join done");

×

998

×

1000

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

×

1001

  SHJoinOperatorInfo* pJoin = pOperator->info;

×

1002

  SExecTaskInfo*      pTaskInfo = pOperator->pTaskInfo;

×

1003

  int32_t             code = TSDB_CODE_SUCCESS;

×

1004

  int32_t             lino = 0;

×

1005

  SSDataBlock*        pRes = pJoin->finBlk;

×

1006

  int64_t             st = 0;

×

1008

  QRY_PARAM_CHECK(pResBlock);

×

1009

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

×

1010

    st = taosGetTimestampUs();

×

1013

  if (pOperator->status == OP_EXEC_DONE) {

×

1014

    pRes->info.rows = 0;

×

1015

    goto _end;

×

1018

  if (!pJoin->keyHashBuilt) {

×

1019

    pJoin->keyHashBuilt = true;

×

1021

    bool queryDone = false;

×

1022

    code = hJoinBuildHash(pOperator, &queryDone);

×

1023

    QUERY_CHECK_CODE(code, lino, _end);

×

1025

    if (queryDone) {

×

1026

      goto _end;

×

1030

  blockDataCleanup(pRes);

×

1032

  if (pJoin->ctx.rowRemains) {

×

1033

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

×

1034

    QUERY_CHECK_CODE(code, lino, _end);

×

1036

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

×

1037

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

×

1038

      QUERY_CHECK_CODE(code, lino, _end);

×

1041

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

×

1042

      *pResBlock = pRes;

×

1043

      return code;

×

1047

  while (true) {

×

1048

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

×

1049

    if (NULL == pBlock) {

×

1050

      hJoinSetDone(pOperator);

×

1051

      break;

×

1054

    pJoin->execInfo.probeBlkNum++;

×

1055

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

×

1057

    code = hJoinPrepareStart(pOperator, pBlock);

×

1058

    QUERY_CHECK_CODE(code, lino, _end);

×

1060

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

×

1061

      continue;

×

1064

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

×

1065

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

×

1066

      QUERY_CHECK_CODE(code, lino, _end);

×

1069

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

×

1070

      break;

×

1074

_end:

×

1075

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

×

1076

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

×

1078

  if (code != TSDB_CODE_SUCCESS) {

×

1079

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

×

1080

    pTaskInfo->code = code;

×

1081

    T_LONG_JMP(pTaskInfo->env, code);

×

1083

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

×

1084

    *pResBlock = pRes;

×

1085

    qDebug("%s %s output %" PRId64 " rows final res", GET_TASKID(pTaskInfo), __func__, pRes->info.rows);

×

1088

  return code;

×

1091

static void destroyHashJoinOperator(void* param) {

×

1092

  SHJoinOperatorInfo* pJoinOperator = (SHJoinOperatorInfo*)param;

×

1093

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

×

1097

  hJoinDestroyKeyHash(&pJoinOperator->pKeyHash);

×

1099

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

×

1100

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

×

1101

  blockDataDestroy(pJoinOperator->finBlk);

×

1102

  pJoinOperator->finBlk = NULL;

×

1103

  taosMemoryFreeClear(pJoinOperator->pResColMap);

×

1104

  taosArrayDestroyEx(pJoinOperator->pRowBufs, hJoinFreeBufPage);

×

1106

  taosMemoryFreeClear(param);

×

1107

×

1109

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

×

1110

  switch (pJoin->joinType) {

×

1111

    case JOIN_TYPE_INNER: {

×

1112

      SNode* pCond = NULL;

×

1113

      if (pJoinNode->pFullOnCond != NULL) {

×

1114

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

×

1115

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

×

1117

        pCond = pJoinNode->pFullOnCond;

×

1118

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

×

1119

        pCond = pJoinNode->node.pConditions;

×

1122

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

×

1123

      break;

×

1125

    case JOIN_TYPE_LEFT:

×

1128

      if (pJoinNode->pFullOnCond != NULL) {

×

1129

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

×

1132

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

×

1133

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

×

1136

      break;

×

1137

    default:

×

1138

      break;

×

1141

  return TSDB_CODE_SUCCESS;

×

1144

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

×

1145

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

×

1146

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

×

1147

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

×

1148

    return (maxRows > limitMaxRows) ? limitMaxRows : maxRows;

×

1151

  return maxRows;

×

1155

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

×

1156

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

×

1157

  if (NULL == pJoin->finBlk) {

×

1158

    QRY_ERR_RET(terrno);

×

1161

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

×

1162

  if (TSDB_CODE_SUCCESS != code) {

×

1163

    QRY_ERR_RET(code);

×

1166

  if (NULL != pJoin->pPreFilter) {

×

1167

    pJoin->midBlk = NULL;

×

1168

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

×

1169

    if (code) {

×

1170

      QRY_ERR_RET(code);

×

1173

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

×

1174

    if (TSDB_CODE_SUCCESS != code) {

×

1175

      QRY_ERR_RET(code);

×

1179

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

×

1180

  return TSDB_CODE_SUCCESS;

×

1183

static int32_t resetHashJoinOperState(SOperatorInfo* pOper) {

×

1184

  SHJoinOperatorInfo* pHjOper = pOper->info;

×

1185

  pHjOper->keyHashBuilt = false;

×

1186

  blockDataCleanup(pHjOper->midBlk);

×

1187

  blockDataCleanup(pHjOper->finBlk);

×

1188

  pOper->status = OP_NOT_OPENED;

×

1190

  pHjOper->execInfo = (SHJoinExecInfo){0};

×

1192

  void*   pIte = NULL;

×

1193

  int32_t iter = 0;

×

1194

  while ((pIte = tSimpleHashIterate(pHjOper->pKeyHash, pIte, &iter)) != NULL) {

×

1195

    SGroupData* pGroup = pIte;

×

1196

    SBufRowInfo* pRow = pGroup->rows;

×

1197

    SBufRowInfo* pNext = NULL;

×

1198

    while (pRow) {

×

1199

      pNext = pRow->next;

×

1200

      taosMemoryFree(pRow);

×

1201

      pRow = pNext;

×

1204

  tSimpleHashCleanup(pHjOper->pKeyHash);

×

1205

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

×

1206

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

×

1207

  if (pHjOper->pKeyHash == NULL) {

×

1208

    return terrno;

×

1210

  taosArrayDestroyEx(pHjOper->pRowBufs, hJoinFreeBufPage);

×

1211

  int32_t code = hJoinInitBufPages(pHjOper);

×

1212

  int64_t limit = pHjOper->ctx.limit;

×

1213

  pHjOper->ctx = (SHJoinCtx){0};

×

1214

  pHjOper->ctx.limit = limit;

×

1215

  return code;

×

1218

int32_t createHashJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream,

×

1220

  QRY_PARAM_CHECK(pOptrInfo);

×

1222

  int32_t             code = TSDB_CODE_SUCCESS;

×

1223

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

×

1224

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

×

1225

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

×

1226

    code = terrno;

×

1227

    goto _return;

×

1230

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

×

1231

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

×

1233

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

×

1235

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

×

1237

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

×

1238

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

×

1240

  hJoinSetBuildAndProbeTable(pInfo, pJoinNode);

×

1242

  HJ_ERR_JRET(hJoinBuildResColsMap(pInfo, pJoinNode));

×

1244

  HJ_ERR_JRET(hJoinInitBufPages(pInfo));

×

1246

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

×

1247

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

×

1248

  if (pInfo->pKeyHash == NULL) {

×

1249

    code = terrno;

×

1250

    goto _return;

×

1253

  HJ_ERR_JRET(hJoinHandleConds(pInfo, pJoinNode, pTaskInfo));

×

1255

  HJ_ERR_JRET(hJoinInitResBlocks(pInfo, pJoinNode));

×

1257

  HJ_ERR_JRET(hJoinSetImplFp(pInfo));

×

1259

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

×

1261

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

×

1262

  setOperatorResetStateFn(pOperator, resetHashJoinOperState);

×

1264

  qDebug("create hash Join operator done");

×

1266

  *pOptrInfo = pOperator;

×

1267

  return TSDB_CODE_SUCCESS;

×

1269

_return:

×

1271

  if (pInfo != NULL) {

×

1272

    destroyHashJoinOperator(pInfo);

×

1274

  destroyOperatorAndDownstreams(pOperator, pDownstream, numOfDownstream);

×

1275

  pTaskInfo->code = code;

×

1276

  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