From 8ee8785dd6fd99c56a95bee105ebe9e108177cad Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 25 Jan 2022 23:08:22 +0800 Subject: [PATCH] [td-11818] refactor, fix bug in select --- source/dnode/vnode/inc/tsdb.h | 1 - source/dnode/vnode/src/tsdb/tsdbRead.c | 70 +++++++++---------------- source/libs/executor/inc/executil.h | 2 +- source/libs/executor/inc/executorimpl.h | 4 +- source/libs/executor/src/executor.c | 10 ++-- source/libs/executor/src/executorimpl.c | 33 +++--------- 6 files changed, 40 insertions(+), 80 deletions(-) diff --git a/source/dnode/vnode/inc/tsdb.h b/source/dnode/vnode/inc/tsdb.h index b11cb6be7a..5f5acc1b05 100644 --- a/source/dnode/vnode/inc/tsdb.h +++ b/source/dnode/vnode/inc/tsdb.h @@ -75,7 +75,6 @@ typedef struct STsdbQueryCond { } STsdbQueryCond; typedef struct { - void *pTable; TSKEY lastKey; uint64_t uid; } STableKeyInfo; diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 4ab364f552..1bca7aea10 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -81,7 +81,6 @@ enum { CHECKINFO_CHOSEN_BOTH = 2 //for update=2(merge case) }; - typedef struct STableCheckInfo { uint64_t tableId; TSKEY lastKey; @@ -258,9 +257,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S for (int32_t j = 0; j < gsize; ++j) { STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(group, j); - STableCheckInfo info = { .lastKey = pKeyInfo->lastKey}; - - info.tableId = pKeyInfo->uid; + STableCheckInfo info = { .lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid}; if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) { if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) { info.lastKey = pTsdbReadHandle->window.skey; @@ -277,12 +274,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S } // TODO group table according to the tag value. -// taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar); -// size_t gsize = taosArrayGetSize(pTableCheckInfo); -// for (int32_t i = 0; i < gsize; ++i) { -// STableCheckInfo* pInfo = (STableCheckInfo*) taosArrayGet(pTableCheckInfo, i); -// } - + taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar); return pTableCheckInfo; } @@ -2605,7 +2597,7 @@ static int32_t getAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) { break; } - STableKeyInfo info = {.pTable = NULL, .lastKey = TSKEY_INITIAL_VAL, uid = id}; + STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id}; taosArrayPush(list, &info); } @@ -3196,7 +3188,7 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) { if (key < lastKey) { key = lastKey; - keyInfo.pTable = pInfo->pTable; +// keyInfo.pTable = pInfo->pTable; keyInfo.lastKey = key; pInfo->lastKey = key; @@ -3210,29 +3202,19 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) { } } - // clear current group, unref unused table - for (int32_t i = 0; i < numOfTables; ++i) { - STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i); - - // keyInfo.pTable may be NULL here. - if (pInfo->pTable != keyInfo.pTable) { -// tsdbUnRefTable(pInfo->pTable); - } - } - // more than one table in each group, only one table left for each group - if (keyInfo.pTable != NULL) { - totalNumOfTable++; - if (taosArrayGetSize(pGroup) == 1) { - // do nothing - } else { - taosArrayClear(pGroup); - taosArrayPush(pGroup, &keyInfo); - } - } else { // mark all the empty groups, and remove it later - taosArrayDestroy(pGroup); - taosArrayPush(emptyGroup, &j); - } +// if (keyInfo.pTable != NULL) { +// totalNumOfTable++; +// if (taosArrayGetSize(pGroup) == 1) { +// // do nothing +// } else { +// taosArrayClear(pGroup); +// taosArrayPush(pGroup, &keyInfo); +// } +// } else { // mark all the empty groups, and remove it later +// taosArrayDestroy(pGroup); +// taosArrayPush(emptyGroup, &j); +// } } // window does not being updated, so set the original @@ -3422,11 +3404,13 @@ void filterPrepare(void* expr, void* param) { } } +#endif static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) { +#if 0 STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param; - STable* pTable1 = ((STableKeyInfo*) p1)->pTable; - STable* pTable2 = ((STableKeyInfo*) p2)->pTable; + STable* pTable1 = ((STableKeyInfo*) p1)->uid; + STable* pTable2 = ((STableKeyInfo*) p2)->uid; for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) { SColIndex* pColIndex = &pTableGroupSupp->pCols[i]; @@ -3474,10 +3458,9 @@ static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *pa return ret; } } - +#endif return 0; } -#endif static int tsdbCheckInfoCompar(const void* key1, const void* key2) { if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) { @@ -3493,10 +3476,9 @@ static int tsdbCheckInfoCompar(const void* key1, const void* key2) { void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTables, TSKEY skey, STableGroupSupporter* pSupp, __ext_compar_fn_t compareFn) { STable* pTable = taosArrayGetP(pTableList, 0); - SArray* g = taosArrayInit(16, sizeof(STableKeyInfo)); - STableKeyInfo info = {.pTable = pTable, .lastKey = skey}; + STableKeyInfo info = {.lastKey = skey}; taosArrayPush(g, &info); for (int32_t i = 1; i < numOfTables; ++i) { @@ -3507,13 +3489,13 @@ void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTable assert(ret == 0 || ret == -1); if (ret == 0) { - STableKeyInfo info1 = {.pTable = *p, .lastKey = skey}; + STableKeyInfo info1 = {.lastKey = skey}; taosArrayPush(g, &info1); } else { taosArrayPush(pGroups, &g); // current group is ended, start a new group g = taosArrayInit(16, sizeof(STableKeyInfo)); - STableKeyInfo info1 = {.pTable = *p, .lastKey = skey}; + STableKeyInfo info1 = {.lastKey = skey}; taosArrayPush(g, &info1); } } @@ -3546,8 +3528,8 @@ SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColInd sup.pTagSchema = pTagSchema->pSchema; sup.pCols = pCols; -// taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn); -// createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn); + taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn); + createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn); } return pTableGroup; diff --git a/source/libs/executor/inc/executil.h b/source/libs/executor/inc/executil.h index 702581edb9..130e46fc4c 100644 --- a/source/libs/executor/inc/executil.h +++ b/source/libs/executor/inc/executil.h @@ -38,7 +38,7 @@ #define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t)) #define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES) -#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.idstr) +#define GET_TASKID(_t) (((SExecTaskInfo*)(_t))->id.str) #define curTimeWindowIndex(_winres) ((_winres)->curIndex) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index f36071f46c..7af2e9f14f 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -239,8 +239,7 @@ typedef struct STaskIdInfo { uint64_t queryId; // this is also a request id uint64_t subplanId; uint64_t templateId; - uint64_t taskId; // this is a subplan id - char *idstr; + char *str; } STaskIdInfo; typedef struct SExecTaskInfo { @@ -639,7 +638,6 @@ int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg); bool isTaskKilled(SExecTaskInfo *pTaskInfo); int32_t checkForQueryBuf(size_t numOfTables); bool checkNeedToCompressQueryCol(SQInfo *pQInfo); -bool doBuildResCheck(SQInfo* pQInfo); void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status); bool onlyQueryTags(STaskAttr* pQueryAttr); diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index a2eb5dc339..a32268bf47 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -18,20 +18,20 @@ #include "executorimpl.h" #include "planner.h" -static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, uint64_t reqId) { +static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) { ASSERT(pOperator != NULL); if (pOperator->operatorType != OP_StreamScan) { if (pOperator->numOfDownstream == 0) { - qError("failed to find stream scan operator to set the input data block, reqId:0x%" PRIx64, reqId); + qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id); return TSDB_CODE_QRY_APP_ERROR; } if (pOperator->numOfDownstream > 1) { // not handle this in join query - qError("join not supported for stream block scan, reqId:0x%" PRIx64, reqId); + qError("join not supported for stream block scan, %s" PRIx64, id); return TSDB_CODE_QRY_APP_ERROR; } - return doSetStreamBlock(pOperator->pDownstream[0], input, reqId); + return doSetStreamBlock(pOperator->pDownstream[0], input, id); } else { SStreamBlockScanInfo* pInfo = pOperator->info; tqReadHandleSetMsg(pInfo->readerHandle, input, 0); @@ -50,7 +50,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) { SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo; - int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, pTaskInfo->id.queryId); + int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, GET_TASKID(pTaskInfo)); if (code != TSDB_CODE_SUCCESS) { qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo)); } else { diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index f4b88f357d..45d1f57e68 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -5146,7 +5146,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) { size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); if (pExchangeInfo->current >= totalSources) { - qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources, + qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources, pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0); return NULL; } @@ -5208,7 +5208,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) { int64_t el = taosGetTimestampUs() - startTs; pExchangeInfo->totalElapsed += el; - qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources, + qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources, pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0); return NULL; } else { @@ -7741,11 +7741,10 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId) { pTaskInfo->cost.created = taosGetTimestampMs(); pTaskInfo->id.queryId = queryId; - pTaskInfo->id.taskId = taskId; char* p = calloc(1, 128); snprintf(p, 128, "TID:0x%"PRIx64" QID:0x%"PRIx64, taskId, queryId); - pTaskInfo->id.idstr = strdup(p); + pTaskInfo->id.str = strdup(p); return pTaskInfo; } @@ -7832,7 +7831,7 @@ static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode, SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo)); - STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = uid}; + STableKeyInfo info = {.lastKey = 0, .uid = uid}; taosArrayPush(pa, &info); taosArrayPush(groupInfo.pGroupList, &pa); } @@ -8827,33 +8826,15 @@ void* freeColumnInfo(SColumnInfo* pColumnInfo, int32_t numOfCols) { } void doDestroyTask(SExecTaskInfo *pTaskInfo) { - qDebug("%s start to free execTask", GET_TASKID(pTaskInfo)); doDestroyTableQueryInfo(&pTaskInfo->tableqinfoGroupInfo); // taosArrayDestroy(pTaskInfo->summary.queryProfEvents); // taosHashCleanup(pTaskInfo->summary.operatorProfResults); + tfree(pTaskInfo->sql); + tfree(pTaskInfo->id.str); qDebug("%s execTask is freed", GET_TASKID(pTaskInfo)); - tfree(pTaskInfo); -} - -bool doBuildResCheck(SQInfo* pQInfo) { - bool buildRes = false; - - pthread_mutex_lock(&pQInfo->lock); - pQInfo->dataReady = QUERY_RESULT_READY; - buildRes = needBuildResAfterQueryComplete(pQInfo); - - // clear qhandle owner, it must be in the secure area. other thread may run ahead before current, after it is - // put into task to be executed. - assert(pQInfo->owner == taosGetSelfPthreadId()); - pQInfo->owner = 0; - - pthread_mutex_unlock(&pQInfo->lock); - - // used in retrieve blocking model. - tsem_post(&pQInfo->ready); - return buildRes; + tfree(pTaskInfo); } static void doSetTagValueToResultBuf(char* output, const char* val, int16_t type, int16_t bytes) { -- GitLab