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

Merge pull request #14030 from taosdata/szhou/feature/sort-group 

feat: order group by group id generated by partition
This commit is contained in:
shenglian-zhou 2022-06-20 20:55:51 +08:00 committed by GitHub
parent 57e6846dae 48d22c359f
commit 0f3fc6cff0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 29 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/libs/executor/inc/executorimpl.h
View File

@ -506,7 +506,8 @@ typedef struct SPartitionOperatorInfo {
SDiskbasedBuf* pBuf; // query result buffer based on blocked-wised disk file SDiskbasedBuf* pBuf; // query result buffer based on blocked-wised disk file
int32_t rowCapacity; // maximum number of rows for each buffer page int32_t rowCapacity; // maximum number of rows for each buffer page
int32_t* columnOffset; // start position for each column data int32_t* columnOffset; // start position for each column data
void* pGroupIter; // group iterator SArray* sortedGroupArray; // SDataGroupInfo sorted by group id
int32_t groupIndex; // group index
int32_t pageIndex; // page index of current group int32_t pageIndex; // page index of current group
SSDataBlock* pUpdateRes; SSDataBlock* pUpdateRes;
SExprSupp scalarSup; SExprSupp scalarSup;

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

@ -586,24 +586,30 @@ static void clearPartitionOperator(SPartitionOperatorInfo* pInfo) {
while( (ite = taosHashIterate(pInfo->pGroupSet, ite)) != NULL ) { while( (ite = taosHashIterate(pInfo->pGroupSet, ite)) != NULL ) {
taosArrayDestroy( ((SDataGroupInfo *)ite)->pPageList); taosArrayDestroy( ((SDataGroupInfo *)ite)->pPageList);
} }
taosHashClear(pInfo->pGroupSet); taosArrayClear(pInfo->sortedGroupArray);
clearDiskbasedBuf(pInfo->pBuf); clearDiskbasedBuf(pInfo->pBuf);
} }
static int compareDataGroupInfo(const void* group1, const void* group2) {
const SDataGroupInfo* pGroupInfo1 = group1;
const SDataGroupInfo* pGroupInfo2 = group2;
return pGroupInfo1->groupId - pGroupInfo2->groupId;
}
static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) { static SSDataBlock* buildPartitionResult(SOperatorInfo* pOperator) {
SPartitionOperatorInfo* pInfo = pOperator->info; SPartitionOperatorInfo* pInfo = pOperator->info;
SDataGroupInfo* pGroupInfo = pInfo->pGroupIter; SDataGroupInfo* pGroupInfo = (pInfo->groupIndex != -1) ? taosArrayGet(pInfo->sortedGroupArray, pInfo->groupIndex) : NULL;
if (pInfo->pGroupIter == NULL || pInfo->pageIndex >= taosArrayGetSize(pGroupInfo->pPageList)) { if (pInfo->groupIndex == -1 || pInfo->pageIndex >= taosArrayGetSize(pGroupInfo->pPageList)) {
// try next group data // try next group data
pInfo->pGroupIter = taosHashIterate(pInfo->pGroupSet, pInfo->pGroupIter); ++pInfo->groupIndex;
if (pInfo->pGroupIter == NULL) { if (pInfo->groupIndex >= taosArrayGetSize(pInfo->sortedGroupArray)) {
doSetOperatorCompleted(pOperator); doSetOperatorCompleted(pOperator);
clearPartitionOperator(pInfo); clearPartitionOperator(pInfo);
return NULL; return NULL;
} }
pGroupInfo = pInfo->pGroupIter; pGroupInfo = taosArrayGet(pInfo->sortedGroupArray, pInfo->groupIndex);
pInfo->pageIndex = 0; pInfo->pageIndex = 0;
} }
@ -657,6 +663,20 @@ static SSDataBlock* hashPartition(SOperatorInfo* pOperator) {
doHashPartition(pOperator, pBlock); doHashPartition(pOperator, pBlock);
} }
SArray* groupArray = taosArrayInit(taosHashGetSize(pInfo->pGroupSet), sizeof(SDataGroupInfo));
void* pGroupIter = NULL;
pGroupIter = taosHashIterate(pInfo->pGroupSet, NULL);
while (pGroupIter != NULL) {
SDataGroupInfo* pGroupInfo = pGroupIter;
taosArrayPush(groupArray, pGroupInfo);
pGroupIter = taosHashIterate(pInfo->pGroupSet, pGroupIter);
}
taosArraySort(groupArray, compareDataGroupInfo);
pInfo->sortedGroupArray = groupArray;
pInfo->groupIndex = -1;
taosHashClear(pInfo->pGroupSet);
pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0; pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
pOperator->status = OP_RES_TO_RETURN; pOperator->status = OP_RES_TO_RETURN;
@ -676,6 +696,7 @@ static void destroyPartitionOperatorInfo(void* param, int32_t numOfOutput) {
taosArrayDestroy(pInfo->pGroupColVals); taosArrayDestroy(pInfo->pGroupColVals);
taosMemoryFree(pInfo->keyBuf); taosMemoryFree(pInfo->keyBuf);
taosArrayDestroy(pInfo->sortedGroupArray);
taosHashCleanup(pInfo->pGroupSet); taosHashCleanup(pInfo->pGroupSet);
taosMemoryFree(pInfo->columnOffset); taosMemoryFree(pInfo->columnOffset);