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

feat: add multi-rows merge join

This commit is contained in:
slzhou@taodata.com 2022-07-27 16:11:32 +08:00
parent 3469117732
commit e1d5971e39
1 changed files with 33 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
source/libs/executor/src/joinoperator.c
View File

@ -146,6 +146,39 @@ static void mergeJoinJoinLeftRight(struct SOperatorInfo* pOperator, SSDataBlock*
} }
} }
typedef struct SRowLocation {
SSDataBlock* pDataBlock;
int32_t pos;
} SRowLocation;
static int32_t mergeJoinGetBlockRowsEqualStart(SSDataBlock* pBlock, int16_t slotId, int32_t startPos,
SArray* pPosArray) {
int32_t numRows = pBlock->info.rows;
ASSERT(startPos < numRows);
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
int32_t i = startPos;
char* pVal = colDataGetData(pCol, i);
for (i = startPos + 1; i < numRows; ++i) {
char* pNextVal = colDataGetData(pCol, i);
if (*(int64_t*)pVal != *(int64_t*)pNextVal) {
break;
}
}
int32_t endPos = i;
SSDataBlock* block = pBlock;
if (endPos - startPos > 1) {
block = blockDataExtractBlock(pBlock, startPos, endPos - startPos);
}
SRowLocation location = {0};
for (int32_t j = startPos; j < endPos; ++j) {
location.pDataBlock = block;
location.pos = j;
taosArrayPush(pPosArray, &location);
}
return 0;
}
static bool mergeJoinGetNextTimestamp(SOperatorInfo* pOperator, int64_t* pLeftTs, int64_t* pRightTs) { static bool mergeJoinGetNextTimestamp(SOperatorInfo* pOperator, int64_t* pLeftTs, int64_t* pRightTs) {
SJoinOperatorInfo* pJoinInfo = pOperator->info; SJoinOperatorInfo* pJoinInfo = pOperator->info;