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

fix: remove group keys from join operator

This commit is contained in:
shenglian zhou 2023-05-18 10:46:19 +08:00
parent 03e77288e8
commit ee16b46112
1 changed files with 18 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -55,12 +55,10 @@ typedef struct SJoinOperatorInfo {
SNode* pTagEqualConditions; SNode* pTagEqualConditions;
SArray* leftTagCols; SArray* leftTagCols;
SArray* leftTagKeys;
char* leftTagKeyBuf; char* leftTagKeyBuf;
int32_t leftTagKeyLen; int32_t leftTagKeyLen;
SArray* rightTagCols; SArray* rightTagCols;
SArray* rightTagKeys;
char* rightTagKeyBuf; char* rightTagKeyBuf;
int32_t rightTagKeyLen; int32_t rightTagKeyLen;
@ -141,27 +139,11 @@ static void extractTagEqualCondCols(SJoinOperatorInfo* pInfo, SOperatorInfo** pD
} }
} }
static int32_t initTagColsGroupkeys(SArray** pGroupColVals, int32_t* keyLen, char** keyBuf, const SArray* pGroupColList) { static int32_t initTagColskeyBuf(int32_t* keyLen, char** keyBuf, const SArray* pGroupColList) {
*pGroupColVals = taosArrayInit(4, sizeof(SGroupKeys));
if ((*pGroupColVals) == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t numOfGroupCols = taosArrayGetSize(pGroupColList); int32_t numOfGroupCols = taosArrayGetSize(pGroupColList);
for (int32_t i = 0; i < numOfGroupCols; ++i) { for (int32_t i = 0; i < numOfGroupCols; ++i) {
SColumn* pCol = (SColumn*)taosArrayGet(pGroupColList, i); SColumn* pCol = (SColumn*)taosArrayGet(pGroupColList, i);
(*keyLen) += pCol->bytes; // actual data + null_flag (*keyLen) += pCol->bytes; // actual data + null_flag
SGroupKeys key = {0};
key.bytes = pCol->bytes;
key.type = pCol->type;
key.isNull = false;
key.pData = taosMemoryCalloc(1, pCol->bytes);
if (key.pData == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
taosArrayPush((*pGroupColVals), &key);
} }
int32_t nullFlagSize = sizeof(int8_t) * numOfGroupCols; int32_t nullFlagSize = sizeof(int8_t) * numOfGroupCols;
@ -175,10 +157,11 @@ static int32_t initTagColsGroupkeys(SArray** pGroupColVals, int32_t* keyLen, cha
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void fillGroupKeyValsFromTagCols(SArray* pCols, SArray* pGroupKeys, SSDataBlock* pBlock, int32_t rowIndex) { static int32_t fillKeyBufFromTagCols(SArray* pCols, SSDataBlock* pBlock, int32_t rowIndex, void* pKey) {
SColumnDataAgg* pColAgg = NULL; SColumnDataAgg* pColAgg = NULL;
size_t numOfGroupCols = taosArrayGetSize(pCols); size_t numOfGroupCols = taosArrayGetSize(pCols);
char* isNull = (char*)pKey;
char* pStart = (char*)pKey + sizeof(int8_t) * numOfGroupCols;
for (int32_t i = 0; i < numOfGroupCols; ++i) { for (int32_t i = 0; i < numOfGroupCols; ++i) {
SColumn* pCol = (SColumn*) taosArrayGet(pCols, i); SColumn* pCol = (SColumn*) taosArrayGet(pCols, i);
@ -193,56 +176,24 @@ static void fillGroupKeyValsFromTagCols(SArray* pCols, SArray* pGroupKeys, SSDat
pColAgg = pBlock->pBlockAgg[pCol->slotId]; // TODO is agg data matched? pColAgg = pBlock->pBlockAgg[pCol->slotId]; // TODO is agg data matched?
} }
SGroupKeys* pkey = taosArrayGet(pGroupKeys, i);
if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) { if (colDataIsNull(pColInfoData, pBlock->info.rows, rowIndex, pColAgg)) {
pkey->isNull = true;
} else {
pkey->isNull = false;
char* val = colDataGetData(pColInfoData, rowIndex);
if (pkey->type == TSDB_DATA_TYPE_JSON) {
if (tTagIsJson(val)) {
terrno = TSDB_CODE_QRY_JSON_IN_GROUP_ERROR;
return;
}
int32_t dataLen = getJsonValueLen(val);
memcpy(pkey->pData, val, dataLen);
} else if (IS_VAR_DATA_TYPE(pkey->type)) {
memcpy(pkey->pData, val, varDataTLen(val));
ASSERT(varDataTLen(val) <= pkey->bytes);
} else {
memcpy(pkey->pData, val, pkey->bytes);
}
}
}
}
static int32_t combineGroupKeysIntoBuf(void* pKey, const SArray* pGroupKeys) {
size_t numOfGroupCols = taosArrayGetSize(pGroupKeys);
char* isNull = (char*)pKey;
char* pStart = (char*)pKey + sizeof(int8_t) * numOfGroupCols;
for (int32_t i = 0; i < numOfGroupCols; ++i) {
SGroupKeys* pkey = taosArrayGet(pGroupKeys, i);
if (pkey->isNull) {
isNull[i] = 1; isNull[i] = 1;
continue;
}
isNull[i] = 0;
if (pkey->type == TSDB_DATA_TYPE_JSON) {
int32_t dataLen = getJsonValueLen(pkey->pData);
memcpy(pStart, (pkey->pData), dataLen);
pStart += dataLen;
} else if (IS_VAR_DATA_TYPE(pkey->type)) {
varDataCopy(pStart, pkey->pData);
pStart += varDataTLen(pkey->pData);
ASSERT(varDataTLen(pkey->pData) <= pkey->bytes);
} else { } else {
memcpy(pStart, pkey->pData, pkey->bytes); isNull[i] = 0;
pStart += pkey->bytes; char* val = colDataGetData(pColInfoData, rowIndex);
if (pCol->type == TSDB_DATA_TYPE_JSON) {
int32_t dataLen = getJsonValueLen(val);
memcpy(pStart, val, dataLen);
pStart += dataLen;
} else if (IS_VAR_DATA_TYPE(pCol->type)) {
varDataCopy(pStart, val);
pStart += varDataTLen(val);
} else {
memcpy(pStart, val, pCol->bytes);
pStart += pCol->bytes;
}
} }
} }
return (int32_t)(pStart - (char*)pKey); return (int32_t)(pStart - (char*)pKey);
} }
@ -312,8 +263,8 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t
pInfo->leftTagCols = taosArrayInit(4, sizeof(SColumn)); pInfo->leftTagCols = taosArrayInit(4, sizeof(SColumn));
pInfo->rightTagCols = taosArrayInit(4, sizeof(SColumn)); pInfo->rightTagCols = taosArrayInit(4, sizeof(SColumn));
extractTagEqualCondCols(pInfo, pDownstream, pInfo->pTagEqualConditions, pInfo->leftTagCols, pInfo->rightTagCols); extractTagEqualCondCols(pInfo, pDownstream, pInfo->pTagEqualConditions, pInfo->leftTagCols, pInfo->rightTagCols);
initTagColsGroupkeys(&pInfo->leftTagKeys, &pInfo->leftTagKeyLen, &pInfo->leftTagKeyBuf, pInfo->leftTagCols); initTagColskeyBuf(&pInfo->leftTagKeyLen, &pInfo->leftTagKeyBuf, pInfo->leftTagCols);
initTagColsGroupkeys(&pInfo->rightTagKeys, &pInfo->rightTagKeyLen, &pInfo->rightTagKeyBuf, pInfo->rightTagCols); initTagColskeyBuf(&pInfo->rightTagKeyLen, &pInfo->rightTagKeyBuf, pInfo->rightTagCols);
} }
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doMergeJoin, NULL, destroyMergeJoinOperator, optrDefaultBufFn, NULL); pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doMergeJoin, NULL, destroyMergeJoinOperator, optrDefaultBufFn, NULL);
code = appendDownstream(pOperator, pDownstream, numOfDownstream); code = appendDownstream(pOperator, pDownstream, numOfDownstream);
@ -341,20 +292,13 @@ void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode) {
pColumn->scale = pColumnNode->node.resType.scale; pColumn->scale = pColumnNode->node.resType.scale;
} }
static void freeGroupKeyData(void* param) {
SGroupKeys* pKey = (SGroupKeys*)param;
taosMemoryFree(pKey->pData);
}
void destroyMergeJoinOperator(void* param) { void destroyMergeJoinOperator(void* param) {
SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param; SJoinOperatorInfo* pJoinOperator = (SJoinOperatorInfo*)param;
if (pJoinOperator->pTagEqualConditions != NULL) { if (pJoinOperator->pTagEqualConditions != NULL) {
taosMemoryFreeClear(pJoinOperator->rightTagKeyBuf); taosMemoryFreeClear(pJoinOperator->rightTagKeyBuf);
taosArrayDestroyEx(pJoinOperator->rightTagKeys, freeGroupKeyData);
taosArrayDestroy(pJoinOperator->rightTagCols); taosArrayDestroy(pJoinOperator->rightTagCols);
taosMemoryFreeClear(pJoinOperator->leftTagKeyBuf); taosMemoryFreeClear(pJoinOperator->leftTagKeyBuf);
taosArrayDestroyEx(pJoinOperator->leftTagKeys, freeGroupKeyData);
taosArrayDestroy(pJoinOperator->leftTagCols); taosArrayDestroy(pJoinOperator->leftTagCols);
} }
nodesDestroyNode(pJoinOperator->pCondAfterMerge); nodesDestroyNode(pJoinOperator->pCondAfterMerge);
@ -482,8 +426,7 @@ static int32_t mergeJoinCreateBuildTable(SJoinOperatorInfo* pInfo, SArray* right
SSHashObj* buildTable = tSimpleHashInit(buildTableCap, hashFn); SSHashObj* buildTable = tSimpleHashInit(buildTableCap, hashFn);
for (int32_t i = 0; i < taosArrayGetSize(rightRowLocations); ++i) { for (int32_t i = 0; i < taosArrayGetSize(rightRowLocations); ++i) {
SRowLocation* rightRow = taosArrayGet(rightRowLocations, i); SRowLocation* rightRow = taosArrayGet(rightRowLocations, i);
fillGroupKeyValsFromTagCols(pInfo->rightTagCols, pInfo->rightTagKeys, rightRow->pDataBlock, rightRow->pos); int32_t keyLen = fillKeyBufFromTagCols(pInfo->rightTagCols, rightRow->pDataBlock, rightRow->pos, pInfo->rightTagKeyBuf);
int32_t keyLen = combineGroupKeysIntoBuf(pInfo->rightTagKeyBuf, pInfo->rightTagKeys);
SArray** ppRows = tSimpleHashGet(buildTable, pInfo->rightTagKeyBuf, keyLen); SArray** ppRows = tSimpleHashGet(buildTable, pInfo->rightTagKeyBuf, keyLen);
if (!ppRows) { if (!ppRows) {
SArray* rows = taosArrayInit(4, sizeof(SRowLocation)); SArray* rows = taosArrayInit(4, sizeof(SRowLocation));
@ -511,8 +454,7 @@ static int32_t mergeJoinLeftRowsRightRows(SOperatorInfo* pOperator, SSDataBlock*
SRowLocation* leftRow = taosArrayGet(leftRowLocations, i); SRowLocation* leftRow = taosArrayGet(leftRowLocations, i);
SArray* pRightRows = NULL; SArray* pRightRows = NULL;
if (rightTableHash != NULL) { if (rightTableHash != NULL) {
fillGroupKeyValsFromTagCols(pJoinInfo->leftTagCols, pJoinInfo->leftTagKeys, leftRow->pDataBlock, leftRow->pos); int32_t keyLen = fillKeyBufFromTagCols(pJoinInfo->leftTagCols, leftRow->pDataBlock, leftRow->pos, pJoinInfo->leftTagKeyBuf);
int32_t keyLen = combineGroupKeysIntoBuf(pJoinInfo->leftTagKeyBuf, pJoinInfo->leftTagKeys);
SArray** ppRightRows = tSimpleHashGet(rightTableHash, pJoinInfo->leftTagKeyBuf, keyLen); SArray** ppRightRows = tSimpleHashGet(rightTableHash, pJoinInfo->leftTagKeyBuf, keyLen);
if (!ppRightRows) { if (!ppRightRows) {
continue; continue;