xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: support left join

This commit is contained in:
dapan1121 2023-12-05 19:27:57 +08:00
parent 4700ebafd1
commit 519622dbf4
2 changed files with 436 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

171
source/libs/executor/inc/mergejoin.h
View File

@ -21,6 +21,18 @@ extern "C" {
typedef SSDataBlock* (*joinImplFp)(SOperatorInfo*); typedef SSDataBlock* (*joinImplFp)(SOperatorInfo*);
typedef enum EJoinTableType {
E_JOIN_TB_BUILD = 1,
E_JOIN_TB_PROBE
} EJoinTableType;
typedef enum EJoinPhase {
E_JOIN_PHASE_RETRIEVE,
E_JOIN_PHASE_SPLIT,
E_JOIN_PHASE_OUTPUT,
E_JOIN_PHASE_
} EJoinPhase;
typedef struct SMJoinColInfo { typedef struct SMJoinColInfo {
int32_t srcSlot; int32_t srcSlot;
int32_t dstSlot; int32_t dstSlot;
@ -33,6 +45,18 @@ typedef struct SMJoinColInfo {
} SMJoinColInfo; } SMJoinColInfo;
typedef struct SMJoinBlkInfo {
bool cloned;
SSDataBlock* pBlk;
void* pNext;
} SMJoinBlkInfo;
typedef struct SMJoinRowInfo {
int64_t blkId;
int32_t rowIdx;
int64_t rowGIdx;
} SMJoinRowInfo;
typedef struct SMJoinTableInfo { typedef struct SMJoinTableInfo {
int32_t downStreamIdx; int32_t downStreamIdx;
SOperatorInfo* downStream; SOperatorInfo* downStream;
@ -56,8 +80,109 @@ typedef struct SMJoinTableInfo {
bool valColExist; bool valColExist;
} SMJoinTableInfo; } SMJoinTableInfo;
typedef struct SMJoinTsJoinCtx {
SMJoinTableCtx* pProbeCtx;
SMJoinTableCtx* pBuildCtx;
int64_t probeRowNum;
int64_t buildRowNum;
int64_t* probeTs;
int64_t* buildTs;
int64_t* probeEndTs;
int64_t* buildEndTs;
bool inSameTsGrp;
bool inDiffTsGrp;
SGrpPairCtx* pLastGrpPairCtx;
SGrpPairCtx currGrpPairCtx;
} SMJoinTsJoinCtx;
typedef struct SBuildGrpCtx {
bool multiBlkGrp;
bool hashJoin;
SSHashObj* pGrpHash;
int32_t grpRowReadIdx;
int32_t grpRowGReadIdx;
int32_t grpRowBeginIdx;
int32_t grpRowNum;
} SBuildGrpCtx;
typedef struct SProbeGrpCtx {
int32_t grpRowReadIdx;
int32_t grpRowBeginIdx;
int32_t grpRowNum;
} SProbeGrpCtx;
typedef struct SGrpPairCtx {
bool sameTsGrp;
bool finishGrp;
SBuildGrpCtx buildGrp;
SProbeGrpCtx probeGrp;
} SGrpPairCtx;
typedef struct SMJoinOutputCtx {
int32_t grpReadIdx;
int32_t grpWriteIdx;
SArray* pGrpList;
} SMJoinOutputCtx;
typedef struct SMJoinTableCtx {
EJoinTableType type;
SMJoinTableInfo* pTbInfo;
bool dsInitDone;
bool dsFetchDone;
int64_t blkCurTs;
int32_t blkRowIdx;
int64_t blkIdx;
int64_t blkNum;
SMJoinBlkInfo* pCurrBlk;
SMJoinBlkInfo* pHeadBlk;
SMJoinBlkInfo* pTailBlk;
} SMJoinTableCtx;
typedef struct SMJoinMergeCtx {
bool hashJoin;
EJoinPhase joinPhase;
int64_t grpCurTs;
SMJoinOutputCtx outputCtx;
SMJoinTsJoinCtx tsJoinCtx;
SMJoinTableCtx buildTbCtx;
SMJoinTableCtx probeTbCtx;
} SMJoinMergeCtx;
typedef struct SMJoinWinCtx {
} SMJoinWinCtx;
typedef struct SMJoinFlowFlags {
bool mergeJoin;
bool windowJoin;
bool preFilter;
bool retrieveAfterBuildDone;
} SMJoinFlowFlags;
typedef struct SMJoinCtx {
SMJoinFlowFlags* pFlags;
union {
SMJoinMergeCtx mergeCtx;
SMJoinWinCtx winCtx;
};
} SMJoinCtx;
typedef struct SMJoinExecInfo {
int64_t buildBlkNum;
int64_t buildBlkRows;
int64_t probeBlkNum;
int64_t probeBlkRows;
int64_t resRows;
int64_t expectRows;
} SMJoinExecInfo;
typedef struct SMJoinOperatorInfo { typedef struct SMJoinOperatorInfo {
SOperatorInfo* pOperator;
int32_t joinType; int32_t joinType;
int32_t subType;
int32_t inputTsOrder; int32_t inputTsOrder;
SMJoinTableInfo tbs[2]; SMJoinTableInfo tbs[2];
SMJoinTableInfo* pBuild; SMJoinTableInfo* pBuild;
@ -65,16 +190,50 @@ typedef struct SMJoinOperatorInfo {
SSDataBlock* pRes; SSDataBlock* pRes;
int32_t pResColNum; int32_t pResColNum;
int8_t* pResColMap; int8_t* pResColMap;
SArray* pRowBufs;
SFilterInfo* pPreFilter; SFilterInfo* pPreFilter;
SFilterInfo* pFinFilter; SFilterInfo* pFinFilter;
SSHashObj* pKeyHash; SMJoinFuncs joinFps;
bool keyHashBuilt; SMJoinCtx ctx;
joinImplFp joinFp; SMJoinExecInfo execInfo;
SHJoinCtx ctx;
SHJoinExecInfo execInfo;
} SMJoinOperatorInfo; } SMJoinOperatorInfo;
#define MJOIN_DOWNSTREAM_NEED_INIT(_pOp) ((_pOp)->pOperatorGetParam && ((SSortMergeJoinOperatorParam*)(_pOp)->pOperatorGetParam->value)->initDownstream)
#define FIN_SAME_TS_GRP() do { \
if (inSameTsGrp) { \
grpPairCtx.sameTsGrp = true; \
grpPairCtx.finishGrp = true; \
grpPairCtx.probeGrp.grpRowBeginIdx = pProbeCtx->blkRowIdx; \
grpPairCtx.probeGrp.grpRowNum = 1; \
inSameTsGrp = false; \
pLastGrpPairCtx = taosArrayPush(pCtx->grpCtx.pGrpList, &grpPairCtx); \
} \
} while (0)
#define PAUSE_SAME_TS_GRP() do { \
if (inSameTsGrp) { \
grpPairCtx.sameTsGrp = true; \
grpPairCtx.finishGrp = false; \
grpPairCtx.probeGrp.grpRowBeginIdx = pProbeCtx->blkRowIdx; \
grpPairCtx.probeGrp.grpRowNum = 1; \
inSameTsGrp = false; \
pLastGrpPairCtx = taosArrayPush(pCtx->grpCtx.pGrpList, &grpPairCtx); \
} \
} while (0)
#define FIN_DIFF_TS_GRP() do { \
if (inDiffTsGrp) { \
grpPairCtx.sameTsGrp = false; \
grpPairCtx.finishGrp = true; \
grpPairCtx.probeGrp.grpRowBeginIdx = pProbeCtx->blkRowIdx; \
grpPairCtx.probeGrp.grpRowNum = 1; \
inDiffTsGrp = false; \
pLastGrpPairCtx = taosArrayPush(pCtx->grpCtx.pGrpList, &grpPairCtx); \
} \
} while (0)
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

318
source/libs/executor/src/mergejoinoperator.c
View File

@ -266,57 +266,280 @@ static int32_t mJoinInitBufPages(SMJoinOperatorInfo* pInfo) {
return mJoinAddPageToBufList(pInfo->pRowBufs); return mJoinAddPageToBufList(pInfo->pRowBufs);
} }
static SSDataBlock* mJoinHanleMergeJoin(SOperatorInfo* pOperator) { static int32_t mJoinDoRetrieve(SMJoinOperatorInfo* pJoin, SMJoinTableCtx* pTbCtx, SMJoinTableInfo* pTbInfo) {
SMJoinOperatorInfo* pJoinInfo = pOperator->info; bool retrieveCont = false;
int32_t code = TSDB_CODE_SUCCESS;
int32_t nrows = pRes->info.rows;
do {
bool asc = (pJoinInfo->inputOrder == TSDB_ORDER_ASC) ? true : false; SSDataBlock* pBlock = getNextBlockFromDownstreamRemain(pJoin->pOperator, pTbInfo->downStreamIdx);
pTbCtx->dsInitDone = true;
while (1) {
int64_t leftTs = 0; if (NULL == pBlock) {
int64_t rightTs = 0; retrieveCont = false;
if (pJoinInfo->rowCtx.rowRemains) { code = (*pJoin->joinFps.handleTbFetchDoneFp)(pJoin, pTbCtx, pTbInfo);
leftTs = pJoinInfo->rowCtx.ts;
rightTs = pJoinInfo->rowCtx.ts;
} else { } else {
bool hasNextTs = mergeJoinGetNextTimestamp(pOperator, &leftTs, &rightTs); code = (*pJoin->joinFps.handleBlkFetchedFp)(pJoin, pTbCtx, pTbInfo, pBlock, &retrieveCont);
if (!hasNextTs) { }
} while (retrieveCont || TSDB_CODE_SUCCESS != code);
return code;
}
static FORCE_INLINE bool mJoinBuildTbNeedRetrieve(SMJoinOperatorInfo* pJoin, SMJoinTableCtx* pBuildCtx) {
if ((!pBuildCtx->grpRetrieved) && (!pBuildCtx->dsFetchDone)) {
return true;
}
return false;
}
static FORCE_INLINE bool mJoinProbeTbNeedRetrieve(SOperatorInfo* pOperator, SMJoinTableCtx* pProbeCtx, SMJoinTableCtx* pBuildCtx) {
SMJoinOperatorInfo* pJoin = pOperator->info;
if (MJOIN_DOWNSTREAM_NEED_INIT(pOperator) && !pBuildCtx->dsInitDone) {
return true;
}
if (((!pProbeCtx->grpRetrieved) && (!pProbeCtx->dsFetchDone)) && (pBuildCtx->grpRetrieved || pJoin->ctx.flags.retrieveAfterBuildDone))) {
return true;
}
return false;
}
static bool mJoinOpenMergeJoin(SOperatorInfo* pOperator, SMJoinOperatorInfo* pJoin) {
SMJoinMergeCtx* pCtx = &pJoin->ctx.mergeCtx;
SMJoinTableCtx* pProbeCtx = &pCtx->probeTbCtx;
SMJoinTableCtx* pBuildCtx = &pCtx->buildTbCtx;
int32_t code = mJoinDoRetrieve(pOperator, pBuildCtx, pJoin->pBuild);
if (TSDB_CODE_SUCCESS != code) {
pOperator->pTaskInfo->code = code;
T_LONG_JMP(pOperator->pTaskInfo->env, code);
}
if (pBuildCtx->dsFetchDone && !pJoin->ctx.flags.retrieveAfterBuildDone) {
return false;
}
code = mJoinDoRetrieve(pOperator, pProbeCtx, pJoin->pProbe);
if (TSDB_CODE_SUCCESS != code) {
pOperator->pTaskInfo->code = code;
T_LONG_JMP(pOperator->pTaskInfo->env, code);
}
return true;
}
static void mJoinInitJoinCtx(SMJoinOperatorInfo* pJoin) {
}
static void mJoinLeftJoinRetrieve(SOperatorInfo* pOperator, SMJoinOperatorInfo* pJoin) {
SMJoinMergeCtx* pCtx = &pJoin->ctx.mergeCtx;
SMJoinTableCtx* pProbeCtx = &pCtx->probeTbCtx;
SMJoinTableCtx* pBuildCtx = &pCtx->buildTbCtx;
int32_t code = mJoinDoRetrieve(pOperator, pProbeCtx, pJoin->pProbe);
if (TSDB_CODE_SUCCESS != code) {
pOperator->pTaskInfo->code = code;
T_LONG_JMP(pOperator->pTaskInfo->env, code);
}
if (NULL == pProbeCtx->pHeadBlk) {
return;
}
code = mJoinDoRetrieve(pJoin, pBuildCtx, pJoin->pBuild);
if (TSDB_CODE_SUCCESS != code) {
pOperator->pTaskInfo->code = code;
T_LONG_JMP(pOperator->pTaskInfo->env, code);
}
pJoin->ctx.mergeCtx.joinPhase = E_JOIN_PHASE_TS_JOIN;
}
static FORCE_INLINE void mJoinLefeJoinUpdateTsJoinCtx(SMJoinTsJoinCtx* pCtx, SMJoinTableCtx* pProbeCtx, SMJoinTableCtx* pBuildCtx) {
pCtx->probeRowNum = pProbeCtx->pCurrBlk->pBlk->info.rows;
pCtx->buildRowNum = pBuildCtx->pCurrBlk->pBlk->info.rows;
SColumnInfoData* probeCol = taosArrayGet(pProbeCtx->pCurrBlk->pBlk, pProbeCtx->pTbInfo->primCol->srcSlot);
SColumnInfoData* buildCol = taosArrayGet(pBuildCtx->pHeadBlk->pBlk, pBuildCtx->pTbInfo->primCol->srcSlot);
pCtx->probeTs = (int64_t*)probeCol->pData;
pCtx->probeEndTs = (int64_t*)probeCol->pData + pCtx->probeRowNum - 1;
pCtx->buildTs = (int64_t*)buildCol->pData;
pCtx->buildEndTs = (int64_t*)buildCol->pData + pCtx->buildRowNum - 1;
}
static bool mJoinMoveToNextProbeTable(SMJoinTsJoinCtx* pCtx, SMJoinTableCtx* pProbeCtx) {
if (NULL == pProbeCtx->pCurrBlk->pNext) {
pProbeCtx->blkIdx++;
return false;
}
pProbeCtx->pCurrBlk = pProbeCtx->pCurrBlk->pNext;
pCtx->probeRowNum = pProbeCtx->pCurrBlk->pBlk->info.rows;
SColumnInfoData* probeCol = taosArrayGet(pProbeCtx->pCurrBlk->pBlk, pProbeCtx->pTbInfo->primCol->srcSlot);
pCtx->probeTs = (int64_t*)probeCol->pData;
pCtx->probeEndTs = (int64_t*)probeCol->pData + pCtx->probeRowNum - 1;
pProbeCtx->blkRowIdx = 0;
return true;
}
static bool mJoinMoveToNextBuildTable(SMJoinOperatorInfo* pJoin, SMJoinTsJoinCtx* pCtx, SMJoinTableCtx* pBuildCtx, SMJoinTableCtx* pProbeCtx) {
bool contLoop = false;
bool res = false;
do {
if (pBuildCtx->pCurrBlk->pNext) {
pBuildCtx->blkIdx++;
return false;
}
pBuildCtx->pCurrBlk = pBuildCtx->pCurrBlk->pNext;
pCtx->buildRowNum = pBuildCtx->pCurrBlk->pBlk->info.rows;
SColumnInfoData* buildCol = taosArrayGet(pBuildCtx->pCurrBlk->pBlk, pBuildCtx->pTbInfo->primCol->srcSlot);
pCtx->buildTs = (int64_t*)buildCol->pData;
pCtx->buildEndTs = (int64_t*)buildCol->pData + pCtx->buildRowNum - 1;
pBuildCtx->blkRowIdx = 0;
do {
if (*pCtx->buildTs > pCtx->probeTs[pProbeCtx->blkRowIdx]) {
mJoinLeftJoinAddBlkToGrp(pJoin, pCtx, pProbeCtx, pBuildCtx);
contLoop = mJoinMoveToNextProbeTable(pCtx, pProbeCtx);
} else if (pCtx->probeTs[pProbeCtx->blkRowIdx] > *pCtx->buildEndTs) {
break; break;
} else {
contLoop = false;
res = true;
}
} while (contLoop);
} while (contLoop);
return res;
}
static void mJoinLeftJoinSplitGrp(SOperatorInfo* pOperator, SMJoinOperatorInfo* pJoin) {
SMJoinTsJoinCtx* pCtx = &pJoin->ctx.mergeCtx.tsJoinCtx;
SMJoinTableCtx* pProbeCtx = &pJoin->ctx.mergeCtx.probeTbCtx;
SMJoinTableCtx* pBuildCtx = &pJoin->ctx.mergeCtx.buildTbCtx;
mJoinLefeJoinUpdateTsJoinCtx(pCtx, pProbeCtx, pBuildCtx);
bool nextRow = false;
for (; pProbeCtx->blkIdx < pProbeCtx->blkNum; mJoinMoveToNextProbeTable(pCtx, pProbeCtx)) {
if (*pCtx->buildTs > *pCtx->probeEndTs) {
mJoinLeftJoinAddBlkToGrp(pJoin, pCtx, pProbeCtx, pBuildCtx);
continue;
} else if (*pCtx->probeTs > *pCtx->buildEndTs) {
if (!mJoinMoveToNextBuildTable(pJoin, pCtx, pBuildCtx, pProbeCtx)) {
break;
//retrieve build
} }
} }
if (leftTs == rightTs) { for (; pProbeCtx->blkRowIdx < pCtx->probeRowNum; ++pProbeCtx->blkRowIdx) {
mergeJoinJoinDownstreamTsRanges(pOperator, leftTs, pRes, &nrows); for (; pBuildCtx->blkIdx < pBuildCtx->blkNum; mJoinMoveToNextBuildTable(pJoin, pCtx, pBuildCtx, pProbeCtx)) {
} else if ((asc && leftTs < rightTs) || (!asc && leftTs > rightTs)) { for (; pBuildCtx->blkRowIdx < pCtx->buildRowNum; ++pBuildCtx->blkRowIdx) {
pJoinInfo->leftPos += 1; if (pCtx->probeTs[pProbeCtx->blkRowIdx] > pCtx->buildTs[pBuildCtx->blkRowIdx]) {
FIN_SAME_TS_GRP();
FIN_DIFF_TS_GRP();
continue;
} else if (pCtx->probeTs[pProbeCtx->blkRowIdx] == pCtx->buildTs[pBuildCtx->blkRowIdx]) {
FIN_DIFF_TS_GRP();
if (pCtx->inSameTsGrp) {
pCtx->currGrpPairCtx.buildGrp.grpRowNum++;
} else {
pCtx->inSameTsGrp = true;
pCtx->currGrpPairCtx.buildGrp.grpRowBeginIdx = pBuildCtx->blkRowIdx;
}
} else {
FIN_SAME_TS_GRP();
if (pCtx->inDiffTsGrp) {
pCtx->currGrpPairCtx.probeGrp.grpRowNum++;
} else {
pCtx->inDiffTsGrp = true;
pCtx->currGrpPairCtx.probeGrp.grpRowBeginIdx = pProbeCtx->blkRowIdx;
}
nextRow = true;
break;
}
}
if (pJoinInfo->leftPos >= pJoinInfo->pLeft->info.rows && pRes->info.rows < pOperator->resultInfo.threshold) { // end of single build table
if (nextRow) {
break;
}
}
// end of all build tables
if (nextRow) {
continue; continue;
} }
} else if ((asc && leftTs > rightTs) || (!asc && leftTs < rightTs)) {
pJoinInfo->rightPos += 1; if (pCtx->inSameTsGrp) {
if (pJoinInfo->rightPos >= pJoinInfo->pRight->info.rows && pRes->info.rows < pOperator->resultInfo.threshold) { PAUSE_SAME_TS_GRP();
continue;
} }
}
// the pDataBlock are always the same one, no need to call this again
pRes->info.rows = nrows;
pRes->info.dataLoad = 1;
pRes->info.scanFlag = MAIN_SCAN;
if (pRes->info.rows >= pOperator->resultInfo.threshold) {
break; break;
} }
// end of single probe table
if (nextRow) {
continue;
}
break;
}
// end of all probe tables
FIN_DIFF_TS_GRP();
pJoin->ctx.mergeCtx.joinPhase = E_JOIN_PHASE_OUTPUT;
return true;
}
static bool mJoinLeftJoinOutput(SOperatorInfo* pOperator, SMJoinOperatorInfo* pJoin) {
}
static SSDataBlock* mJoinDoLeftJoin(SOperatorInfo* pOperator, SMJoinOperatorInfo* pJoin) {
bool contLoop = false;
do {
switch (pJoin->ctx.mergeCtx.joinPhase) {
case E_JOIN_PHASE_RETRIEVE:
contLoop = mJoinLeftJoinRetrieve(pOperator, pJoin);
break;
case E_JOIN_PHASE_SPLIT:
contLoop = mJoinLeftJoinSplitGrp(pOperator, pJoin);
break;
case E_JOIN_PHASE_OUTPUT:
contLoop = mJoinLeftJoinOutput(pOperator, pJoin);
break;
}
} while (contLoop);
}
static SSDataBlock* mJoinHanleMergeJoin(SOperatorInfo* pOperator) {
SMJoinOperatorInfo* pJoin = pOperator->info;
SMJoinMergeCtx* pCtx = &pJoin->ctx.mergeCtx;
SMJoinTableCtx* pProbeCtx = &pCtx->probeTbCtx;
SMJoinTableCtx* pBuildCtx = &pCtx->buildTbCtx;
int32_t code = TSDB_CODE_SUCCESS;
SSDataBlock* pBlock = NULL;
while (true) {
pBlock = (*pJoin->joinFp)(pOperator, pJoin, pCtx, pBuildCtx, pProbeCtx);
if (pBlock && pBlock->info.rows > 0) {
return pBlock;
}
} }
} }
SSDataBlock* mJoinMainProcess(struct SOperatorInfo* pOperator) { SSDataBlock* mJoinMainProcess(struct SOperatorInfo* pOperator) {
SMJoinOperatorInfo* pJoinInfo = pOperator->info; SMJoinOperatorInfo* pJoin = pOperator->info;
if (pOperator->status == OP_EXEC_DONE) { if (pOperator->status == OP_EXEC_DONE) {
if (NULL == pOperator->pDownstreamGetParams || NULL == pOperator->pDownstreamGetParams[0] || NULL == pOperator->pDownstreamGetParams[1]) { if (NULL == pOperator->pDownstreamGetParams || NULL == pOperator->pDownstreamGetParams[0] || NULL == pOperator->pDownstreamGetParams[1]) {
qDebug("%s total merge join res rows:%" PRId64, GET_TASKID(pOperator->pTaskInfo), pJoinInfo->resRows); qDebug("%s total merge join res rows:%" PRId64, GET_TASKID(pOperator->pTaskInfo), pJoin->resRows);
return NULL; return NULL;
} else { } else {
resetMergeJoinOperator(pOperator); resetMergeJoinOperator(pOperator);
@ -329,20 +552,18 @@ SSDataBlock* mJoinMainProcess(struct SOperatorInfo* pOperator) {
st = taosGetTimestampUs(); st = taosGetTimestampUs();
} }
SSDataBlock* pRes = pJoinInfo->pRes; SSDataBlock* pBlock = NULL;
blockDataCleanup(pRes); //blockDataCleanup(pJoin->pRes);
while (true) { while (true) {
int32_t numOfRowsBefore = pRes->info.rows; pBlock = (*pJoin->joinFp)(pOperator);
mJoinImpl(pOperator, pRes); if (NULL == pBlock) {
int32_t numOfNewRows = pRes->info.rows - numOfRowsBefore;
if (numOfNewRows == 0) {
break; break;
} }
if (pJoinInfo->pFinFilter != NULL) { if (pJoin->pFinFilter != NULL) {
doFilter(pRes, pJoinInfo->pFinFilter, NULL); doFilter(pBlock, pJoin->pFinFilter, NULL);
} }
if (pRes->info.rows > 0 || pOperator->status == OP_EXEC_DONE) { if (pBlock->info.rows > 0 || pOperator->status == OP_EXEC_DONE) {
break; break;
} }
} }
@ -351,12 +572,12 @@ SSDataBlock* mJoinMainProcess(struct SOperatorInfo* pOperator) {
pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0; pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
} }
if (pRes->info.rows > 0) { if (pBlock->info.rows > 0) {
pJoinInfo->resRows += pRes->info.rows; pJoin->resRows += pBlock->info.rows;
qDebug("%s merge join returns res rows:%" PRId64, GET_TASKID(pOperator->pTaskInfo), pRes->info.rows); qDebug("%s merge join returns res rows:%" PRId64, GET_TASKID(pOperator->pTaskInfo), pBlock->info.rows);
return pRes; return pBlock;
} else { } else {
qDebug("%s total merge join res rows:%" PRId64, GET_TASKID(pOperator->pTaskInfo), pJoinInfo->resRows); qDebug("%s total merge join res rows:%" PRId64, GET_TASKID(pOperator->pTaskInfo), pJoin->resRows);
return NULL; return NULL;
} }
} }
@ -374,6 +595,7 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
goto _error; goto _error;
} }
pInfo->pOperator = pOperator;
code = mJoinInitDownstreamInfo(pInfo, pDownstream, numOfDownstream, newDownstreams); code = mJoinInitDownstreamInfo(pInfo, pDownstream, numOfDownstream, newDownstreams);
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
goto _error; goto _error;
@ -391,6 +613,8 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
mJoinSetBuildAndProbeTable(pInfo, pJoinNode); mJoinSetBuildAndProbeTable(pInfo, pJoinNode);
mJoinInitJoinCtx(pInfo);
code = mJoinBuildResColMap(pInfo, pJoinNode); code = mJoinBuildResColMap(pInfo, pJoinNode);
if (code) { if (code) {
goto _error; goto _error;