From c2fa9ce29c37c5099abfca7c5f52b3de2727d6cf Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 24 Aug 2021 17:20:00 +0800 Subject: [PATCH] [td-6260]fix the bug found by regression test. --- src/client/src/tscGlobalmerge.c | 38 ++++++++++++++++++++++++++++----- src/client/src/tscSQLParser.c | 15 ++++++++++--- src/client/src/tscUtil.c | 5 ++++- src/query/inc/qExecutor.h | 32 +++++++++++++++------------ src/query/inc/qTableMeta.h | 1 + src/query/src/qExecutor.c | 34 +++++++++++++++++------------ 6 files changed, 88 insertions(+), 37 deletions(-) diff --git a/src/client/src/tscGlobalmerge.c b/src/client/src/tscGlobalmerge.c index ed99fcbabf..3669d84459 100644 --- a/src/client/src/tscGlobalmerge.c +++ b/src/client/src/tscGlobalmerge.c @@ -1021,7 +1021,13 @@ static void ensureOutputBuf(SSLimitOperatorInfo * pInfo, SSDataBlock *pResultBlo } } -static void doSlimitImpl(SOperatorInfo* pOperator, SSLimitOperatorInfo* pInfo, SSDataBlock* pBlock) { +enum { + BLOCK_NEW_GROUP = 1, + BLOCK_NO_GROUP = 2, + BLOCK_SAME_GROUP = 3, +}; + +static int32_t doSlimitImpl(SOperatorInfo* pOperator, SSLimitOperatorInfo* pInfo, SSDataBlock* pBlock) { int32_t rowIndex = 0; while (rowIndex < pBlock->info.rows) { @@ -1030,12 +1036,12 @@ static void doSlimitImpl(SOperatorInfo* pOperator, SSLimitOperatorInfo* pInfo, S bool samegroup = true; if (pInfo->hasPrev) { for (int32_t i = 0; i < numOfCols; ++i) { - SColIndex * pIndex = taosArrayGet(pInfo->orderColumnList, i); + SColIndex *pIndex = taosArrayGet(pInfo->orderColumnList, i); SColumnInfoData *pColInfoData = taosArrayGet(pBlock->pDataBlock, pIndex->colIndex); SColumnInfo *pColInfo = &pColInfoData->info; - char * d = rowIndex * pColInfo->bytes + (char *)pColInfoData->pData; + char *d = rowIndex * pColInfo->bytes + (char *)pColInfoData->pData; int32_t ret = columnValueAscendingComparator(pInfo->prevRow[i], d, pColInfo->type, pColInfo->bytes); if (ret != 0) { // it is a new group samegroup = false; @@ -1063,10 +1069,17 @@ static void doSlimitImpl(SOperatorInfo* pOperator, SSLimitOperatorInfo* pInfo, S if (pInfo->slimit.limit >= 0 && pInfo->groupTotal >= pInfo->slimit.limit) { setQueryStatus(pOperator->pRuntimeEnv, QUERY_COMPLETED); pOperator->status = OP_EXEC_DONE; - return; + return BLOCK_NO_GROUP; } pInfo->groupTotal += 1; + + // data in current group not allowed, return if current result does not belong to the previous group.And there + // are results exists in current SSDataBlock + if (!pInfo->multigroupResult && !samegroup && pInfo->pRes->info.rows > 0) { + return BLOCK_NEW_GROUP; + } + doHandleDataInCurrentGroup(pInfo, pBlock, rowIndex); } else { // handle the offset in the same group @@ -1081,6 +1094,8 @@ static void doSlimitImpl(SOperatorInfo* pOperator, SSLimitOperatorInfo* pInfo, S rowIndex += 1; } + + return BLOCK_SAME_GROUP; } SSDataBlock* doSLimit(void* param, bool* newgroup) { @@ -1092,6 +1107,14 @@ SSDataBlock* doSLimit(void* param, bool* newgroup) { SSLimitOperatorInfo *pInfo = pOperator->info; pInfo->pRes->info.rows = 0; + if (pInfo->pPrevBlock != NULL) { + ensureOutputBuf(pInfo, pInfo->pRes, pInfo->pPrevBlock->info.rows); + int32_t ret = doSlimitImpl(pOperator, pInfo, pInfo->pPrevBlock); + assert(ret != BLOCK_NEW_GROUP); + + pInfo->pPrevBlock = NULL; + } + assert(pInfo->currentGroupOffset >= 0); while(1) { @@ -1104,7 +1127,12 @@ SSDataBlock* doSLimit(void* param, bool* newgroup) { } ensureOutputBuf(pInfo, pInfo->pRes, pBlock->info.rows); - doSlimitImpl(pOperator, pInfo, pBlock); + int32_t ret = doSlimitImpl(pOperator, pInfo, pBlock); + if (ret == BLOCK_NEW_GROUP) { + pInfo->pPrevBlock = pBlock; + return pInfo->pRes; + } + if (pOperator->status == OP_EXEC_DONE) { return pInfo->pRes->info.rows == 0 ? NULL : pInfo->pRes; } diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index ad5e78dba6..6e9506c576 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -925,7 +925,6 @@ int32_t tscValidateSqlInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) { pQueryInfo = pCmd->active; pQueryInfo->pUdfInfo = pUdfInfo; pQueryInfo->udfCopy = true; - } } @@ -8696,6 +8695,7 @@ static int32_t doValidateSubquery(SSqlNode* pSqlNode, int32_t index, SSqlObj* pS if (taosArrayGetSize(subInfo->pSubquery) >= 2) { return invalidOperationMsg(msgBuf, "not support union in subquery"); } + SQueryInfo* pSub = calloc(1, sizeof(SQueryInfo)); tscInitQueryInfo(pSub); @@ -8713,12 +8713,12 @@ static int32_t doValidateSubquery(SSqlNode* pSqlNode, int32_t index, SSqlObj* pS return code; } - // create dummy table meta info STableMetaInfo* pTableMetaInfo1 = calloc(1, sizeof(STableMetaInfo)); if (pTableMetaInfo1 == NULL) { return TSDB_CODE_TSC_OUT_OF_MEMORY; } + pTableMetaInfo1->pTableMeta = extractTempTableMetaFromSubquery(pSub); pTableMetaInfo1->tableMetaCapacity = tscGetTableMetaSize(pTableMetaInfo1->pTableMeta); @@ -8802,7 +8802,7 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf // check if there is 3 level select SRelElementPair* subInfo = taosArrayGet(pSqlNode->from->list, i); SSqlNode* p = taosArrayGetP(subInfo->pSubquery, 0); - if (p->from->type == SQL_NODE_FROM_SUBQUERY){ + if (p->from->type == SQL_NODE_FROM_SUBQUERY) { return invalidOperationMsg(tscGetErrorMsgPayload(pCmd), msg9); } @@ -8895,6 +8895,15 @@ int32_t validateSqlNode(SSqlObj* pSql, SSqlNode* pSqlNode, SQueryInfo* pQueryInf } } + // disable group result mixed up if interval/session window query exists. + if (isTimeWindowQuery(pQueryInfo)) { + size_t num = taosArrayGetSize(pQueryInfo->pUpstream); + for(int32_t i = 0; i < num; ++i) { + SQueryInfo* pUp = taosArrayGetP(pQueryInfo->pUpstream, i); + pUp->multigroupResult = false; + } + } + // parse the having clause in the first place int32_t joinQuery = (pSqlNode->from != NULL && taosArrayGetSize(pSqlNode->from->list) > 1); if (validateHavingClause(pQueryInfo, pSqlNode->pHaving, pCmd, pSqlNode->pSelNodeList, joinQuery, timeWindowQuery) != diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 591a6bba34..3f737d1589 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -3128,6 +3128,7 @@ void tscInitQueryInfo(SQueryInfo* pQueryInfo) { pQueryInfo->slimit.offset = 0; pQueryInfo->pUpstream = taosArrayInit(4, POINTER_BYTES); pQueryInfo->window = TSWINDOW_INITIALIZER; + pQueryInfo->multigroupResult = true; } int32_t tscAddQueryInfo(SSqlCmd* pCmd) { @@ -3139,7 +3140,6 @@ int32_t tscAddQueryInfo(SSqlCmd* pCmd) { } tscInitQueryInfo(pQueryInfo); - pQueryInfo->msg = pCmd->payload; // pointer to the parent error message buffer if (pCmd->pQueryInfo == NULL) { @@ -3222,6 +3222,7 @@ int32_t tscQueryInfoCopy(SQueryInfo* pQueryInfo, const SQueryInfo* pSrc) { pQueryInfo->window = pSrc->window; pQueryInfo->sessionWindow = pSrc->sessionWindow; pQueryInfo->pTableMetaInfo = NULL; + pQueryInfo->multigroupResult = pSrc->multigroupResult; pQueryInfo->bufLen = pSrc->bufLen; pQueryInfo->orderProjectQuery = pSrc->orderProjectQuery; @@ -3623,6 +3624,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, __async_cb_func_t pNewQueryInfo->pTableMetaInfo = NULL; pNewQueryInfo->bufLen = pQueryInfo->bufLen; pNewQueryInfo->distinct = pQueryInfo->distinct; + pNewQueryInfo->multigroupResult = pQueryInfo->multigroupResult; pNewQueryInfo->buf = malloc(pQueryInfo->bufLen); if (pNewQueryInfo->buf == NULL) { @@ -4736,6 +4738,7 @@ int32_t tscCreateQueryFromQueryInfo(SQueryInfo* pQueryInfo, SQueryAttr* pQueryAt pQueryAttr->distinct = pQueryInfo->distinct; pQueryAttr->sw = pQueryInfo->sessionWindow; pQueryAttr->stateWindow = pQueryInfo->stateWindow; + pQueryAttr->multigroupResult = pQueryInfo->multigroupResult; pQueryAttr->numOfCols = numOfCols; pQueryAttr->numOfOutput = numOfOutput; diff --git a/src/query/inc/qExecutor.h b/src/query/inc/qExecutor.h index 611c235c86..f07aac4b93 100644 --- a/src/query/inc/qExecutor.h +++ b/src/query/inc/qExecutor.h @@ -223,6 +223,7 @@ typedef struct SQueryAttr { bool distinct; // distinct query or not bool stateWindow; // window State on sub/normal table bool createFilterOperator; // if filter operator is needed + bool multigroupResult; // multigroup result can exist in one SSDataBlock int32_t interBufSize; // intermediate buffer sizse int32_t havingNum; // having expr number @@ -469,19 +470,21 @@ typedef struct SLimitOperatorInfo { } SLimitOperatorInfo; typedef struct SSLimitOperatorInfo { - int64_t groupTotal; - int64_t currentGroupOffset; - - int64_t rowsTotal; - int64_t currentOffset; - SLimitVal limit; - SLimitVal slimit; - - char **prevRow; - SArray *orderColumnList; - bool hasPrev; - bool ignoreCurrentGroup; + int64_t groupTotal; + int64_t currentGroupOffset; + + int64_t rowsTotal; + int64_t currentOffset; + SLimitVal limit; + SLimitVal slimit; + + char **prevRow; + SArray *orderColumnList; + bool hasPrev; + bool ignoreCurrentGroup; + bool multigroupResult; SSDataBlock *pRes; // result buffer + SSDataBlock *pPrevBlock; int64_t capacity; int64_t threshold; } SSLimitOperatorInfo; @@ -497,6 +500,7 @@ typedef struct SFillOperatorInfo { int64_t totalInputRows; void **p; SSDataBlock *existNewGroupBlock; + bool multigroupResult; } SFillOperatorInfo; typedef struct SGroupbyOperatorInfo { @@ -582,7 +586,7 @@ SOperatorInfo* createLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorI SOperatorInfo* createTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); SOperatorInfo* createAllTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); SOperatorInfo* createSWindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); -SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); +SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, bool multigroupResult); SOperatorInfo* createGroupbyOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); @@ -594,7 +598,7 @@ SOperatorInfo* createMultiwaySortOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SEx int32_t numOfRows, void* merger); SOperatorInfo* createGlobalAggregateOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* param, SArray* pUdfInfo, bool groupResultMixedUp); SOperatorInfo* createStatewindowOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput); -SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* merger); +SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* merger, bool multigroupResult); SOperatorInfo* createFilterOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SColumnInfo* pCols, int32_t numOfFilter); diff --git a/src/query/inc/qTableMeta.h b/src/query/inc/qTableMeta.h index d6b04b0330..746c5f8569 100644 --- a/src/query/inc/qTableMeta.h +++ b/src/query/inc/qTableMeta.h @@ -165,6 +165,7 @@ typedef struct SQueryInfo { bool orderProjectQuery; bool stateWindow; bool globalMerge; + bool multigroupResult; } SQueryInfo; /** diff --git a/src/query/src/qExecutor.c b/src/query/src/qExecutor.c index 651cb8f07d..2ec87075c6 100644 --- a/src/query/src/qExecutor.c +++ b/src/query/src/qExecutor.c @@ -2247,7 +2247,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf case OP_Fill: { SOperatorInfo* pInfo = pRuntimeEnv->proot; - pRuntimeEnv->proot = createFillOperatorInfo(pRuntimeEnv, pInfo, pInfo->pExpr, pInfo->numOfOutput); + pRuntimeEnv->proot = createFillOperatorInfo(pRuntimeEnv, pInfo, pInfo->pExpr, pInfo->numOfOutput, pQueryAttr->multigroupResult); break; } @@ -2257,16 +2257,20 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf } case OP_GlobalAggregate: { // If fill operator exists, the result rows of different group can not be in the same SSDataBlock. - bool groupResultMixedUp = (pQueryAttr->fillType == TSDB_FILL_NONE); + bool multigroupResult = pQueryAttr->multigroupResult; + if (pQueryAttr->multigroupResult) { + multigroupResult = (pQueryAttr->fillType == TSDB_FILL_NONE); + } + pRuntimeEnv->proot = createGlobalAggregateOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr3, - pQueryAttr->numOfExpr3, merger, pQueryAttr->pUdfInfo, groupResultMixedUp); + pQueryAttr->numOfExpr3, merger, pQueryAttr->pUdfInfo, multigroupResult); break; } case OP_SLimit: { int32_t num = pRuntimeEnv->proot->numOfOutput; SExprInfo* pExpr = pRuntimeEnv->proot->pExpr; - pRuntimeEnv->proot = createSLimitOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pExpr, num, merger); + pRuntimeEnv->proot = createSLimitOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pExpr, num, merger, pQueryAttr->multigroupResult); break; } @@ -6345,7 +6349,7 @@ static void doHandleRemainBlockFromNewGroup(SFillOperatorInfo *pInfo, SQueryRunt if (taosFillHasMoreResults(pInfo->pFillInfo)) { *newgroup = false; doFillTimeIntervalGapsInResults(pInfo->pFillInfo, pInfo->pRes, (int32_t)pRuntimeEnv->resultInfo.capacity, pInfo->p); - if (pInfo->pRes->info.rows > pRuntimeEnv->resultInfo.threshold) { + if (pInfo->pRes->info.rows > pRuntimeEnv->resultInfo.threshold || (!pInfo->multigroupResult)) { return; } } @@ -6377,7 +6381,7 @@ static SSDataBlock* doFill(void* param, bool* newgroup) { SQueryRuntimeEnv *pRuntimeEnv = pOperator->pRuntimeEnv; doHandleRemainBlockFromNewGroup(pInfo, pRuntimeEnv, newgroup); - if (pInfo->pRes->info.rows > pRuntimeEnv->resultInfo.threshold) { + if (pInfo->pRes->info.rows > pRuntimeEnv->resultInfo.threshold || (!pInfo->multigroupResult && pInfo->pRes->info.rows > 0)) { return pInfo->pRes; } // if (taosFillHasMoreResults(pInfo->pFillInfo)) { @@ -6414,8 +6418,8 @@ static SSDataBlock* doFill(void* param, bool* newgroup) { pInfo->existNewGroupBlock = pBlock; *newgroup = false; - // fill the previous group data block - // before handle a new data block, close the fill operation for previous group data block + // Fill the previous group data block, before handle the data block of new group. + // Close the fill operation for previous group data block taosFillSetStartInfo(pInfo->pFillInfo, 0, pRuntimeEnv->pQueryAttr->window.ekey); } else { if (pBlock == NULL) { @@ -6436,8 +6440,9 @@ static SSDataBlock* doFill(void* param, bool* newgroup) { // current group has no more result to return if (pInfo->pRes->info.rows > 0) { - // the result in current group not reach the threshold of output result, continue - if (pInfo->pRes->info.rows > pRuntimeEnv->resultInfo.threshold || pBlock == NULL) { + // 1. The result in current group not reach the threshold of output result, continue + // 2. If multiple group results existing in one SSDataBlock is not allowed, return immediately + if (pInfo->pRes->info.rows > pRuntimeEnv->resultInfo.threshold || pBlock == NULL || (!pInfo->multigroupResult)) { return pInfo->pRes; } @@ -6932,10 +6937,10 @@ SOperatorInfo* createGroupbyOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperato return pOperator; } -SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, - int32_t numOfOutput) { +SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, bool multigroupResult) { SFillOperatorInfo* pInfo = calloc(1, sizeof(SFillOperatorInfo)); pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pRuntimeEnv->resultInfo.capacity); + pInfo->multigroupResult = multigroupResult; { SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr; @@ -6971,7 +6976,7 @@ SOperatorInfo* createFillOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorIn return pOperator; } -SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* pMerger) { +SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, void* pMerger, bool multigroupResult) { SSLimitOperatorInfo* pInfo = calloc(1, sizeof(SSLimitOperatorInfo)); SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr; @@ -6982,7 +6987,8 @@ SOperatorInfo* createSLimitOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperator pInfo->capacity = pRuntimeEnv->resultInfo.capacity; pInfo->threshold = pInfo->capacity * 0.8; pInfo->currentOffset = pQueryAttr->limit.offset; - pInfo->currentGroupOffset = pQueryAttr->slimit.offset; + pInfo->currentGroupOffset = pQueryAttr->slimit.offset; + pInfo->multigroupResult= multigroupResult; // TODO refactor int32_t len = 0; -- GitLab