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

refactor: opt filter perf.

This commit is contained in:
Haojun Liao 2022-09-20 13:40:40 +08:00
parent bf33be4207
commit b567a1ebc3
3 changed files with 52 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
include/libs/scalar/filter.h
View File

@ -31,13 +31,17 @@ enum {
FLT_OPTION_NEED_UNIQE = 4,
};
#define FILTER_RESULT_ALL_QUALIFIED 0x1
#define FILTER_RESULT_NONE_QUALIFIED 0x2
#define FILTER_RESULT_PARTIAL_QUALIFIED 0x3
typedef struct SFilterColumnParam {
int32_t numOfCols;
SArray *pDataBlock;
} SFilterColumnParam;
extern int32_t filterInitFromNode(SNode *pNode, SFilterInfo **pinfo, uint32_t options);
extern bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t **p, SColumnDataAgg *statis, int16_t numOfCols);
extern bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t **p, SColumnDataAgg *statis, int16_t numOfCols, int32_t* pResultStatus);
extern int32_t filterSetDataFromSlotId(SFilterInfo *info, void *param);
extern int32_t filterSetDataFromColId(SFilterInfo *info, void *param);
extern int32_t filterGetTimeRange(SNode *pNode, STimeWindow *win, bool *isStrict);

38
source/libs/executor/src/executorimpl.c
View File

@ -1115,7 +1115,7 @@ void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numO
}
}
static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep);
static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep, int32_t status);
void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColMatchInfo) {
if (pFilterNode == NULL || pBlock->info.rows == 0) {
@ -1143,19 +1143,19 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColM
int8_t* rowRes = NULL;
// todo the keep seems never to be True??
bool keep = filterExecute(filter, pBlock, &rowRes, NULL, param1.numOfCols);
int32_t status = 0;
bool keep = filterExecute(filter, pBlock, &rowRes, NULL, param1.numOfCols, &status);
filterFreeInfo(filter);
int64_t st3 = taosGetTimestampUs();
pError("do filter, el: %d us", st3-st2);
extractQualifiedTupleByFilterResult(pBlock, rowRes, keep);
extractQualifiedTupleByFilterResult(pBlock, rowRes, keep, status);
int64_t st4 = taosGetTimestampUs();
pError("extract result filter, el: %d us", st4-st3);
if (pColMatchInfo != NULL) {
for (int32_t i = 0; i < taosArrayGetSize(pColMatchInfo); ++i) {
SColMatchInfo* pInfo = taosArrayGet(pColMatchInfo, i);
@ -1172,13 +1172,34 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, const SArray* pColM
taosMemoryFree(rowRes);
}
void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep) {
void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowRes, bool keep, int32_t status) {
if (keep) {
return;
}
if (rowRes != NULL) {
int32_t totalRows = pBlock->info.rows;
int32_t totalRows = pBlock->info.rows;
if (status == FILTER_RESULT_ALL_QUALIFIED) {
SSDataBlock* px = createOneDataBlock(pBlock, true);
size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pSrc = taosArrayGet(px->pDataBlock, i);
SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
// it is a reserved column for scalar function, and no data in this column yet.
if (pDst->pData == NULL || pSrc->pData == NULL) {
continue;
}
colInfoDataCleanup(pDst, pBlock->info.rows);
colDataAssign(pDst, pSrc, totalRows, NULL);
}
blockDataDestroy(px); // fix memory leak
} else if (status == FILTER_RESULT_NONE_QUALIFIED) {
pBlock->info.rows = 0;
} else {
SSDataBlock* px = createOneDataBlock(pBlock, true);
size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
@ -1214,9 +1235,6 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR
}
blockDataDestroy(px); // fix memory leak
} else {
// do nothing
pBlock->info.rows = 0;
}
}

23
source/libs/scalar/src/filter.c
View File

@ -4026,8 +4026,9 @@ _return:
FLT_RET(code);
}
bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols) {
bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols, int32_t* pResultStatus) {
if (NULL == info) {
*pResultStatus = FILTER_RESULT_ALL_QUALIFIED;
return false;
}
@ -4051,13 +4052,27 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnData
taosMemoryFree(output.columnData);
taosArrayDestroy(pList);
int32_t numOfQualified = 0;
for(int32_t i = 0; i < output.numOfRows; ++i) {
if ((*p)[i] == 1) {
++numOfQualified;
}
}
if (numOfQualified == output.numOfRows) {
*pResultStatus = FILTER_RESULT_ALL_QUALIFIED;
} else if (numOfQualified == 0) {
*pResultStatus = FILTER_RESULT_NONE_QUALIFIED;
} else {
*pResultStatus = FILTER_RESULT_PARTIAL_QUALIFIED;
}
return false;
} else {
return (*info->func)(info, pSrc->info.rows, p, statis, numOfCols);
}
return (*info->func)(info, pSrc->info.rows, p, statis, numOfCols);
}
typedef struct SClassifyConditionCxt {
bool hasPrimaryKey;
bool hasTagIndexCol;