diff --git a/include/common/tcommon.h b/include/common/tcommon.h index 9e3ad42a82fe779bc507417d84718b342a98a34e..0ff13963c069025da6e2143bfcf4a8fdfc46ce7b 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -219,6 +219,16 @@ typedef struct { #define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP) +#define SORT_QSORT_T 0x1 +#define SORT_SPILLED_MERGE_SORT_T 0x2 +typedef struct SSortExecInfo { + int32_t sortMethod; + int32_t sortBuffer; + int32_t loops; // loop count + int32_t writeBytes; // write io bytes + int32_t readBytes; // read io bytes +} SSortExecInfo; + #ifdef __cplusplus } #endif diff --git a/include/common/tdatablock.h b/include/common/tdatablock.h index db8644ecfed50f354e61ff20b424f93dc559f8d7..e8fe47a4621828a3c62fb94b1c516947791f7fdb 100644 --- a/include/common/tdatablock.h +++ b/include/common/tdatablock.h @@ -198,7 +198,7 @@ void colDataTrim(SColumnInfoData* pColumnInfoData); size_t blockDataGetNumOfCols(const SSDataBlock* pBlock); size_t blockDataGetNumOfRows(const SSDataBlock* pBlock); -int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc, SArray* pIndexMap); +int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc); int32_t blockDataSplitRows(SSDataBlock* pBlock, bool hasVarCol, int32_t startIndex, int32_t* stopIndex, int32_t pageSize); int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock); diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 51bcd05ea12a32eb9ac94d1aadea72c37ab531bb..816fdab3d24aa848877f4af8f0444812cfadc140 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -361,19 +361,13 @@ int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex) return 0; } -// if pIndexMap = NULL, merger one column by on column -int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc, SArray* pIndexMap) { +int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc) { assert(pSrc != NULL && pDest != NULL); int32_t capacity = pDest->info.capacity; for (int32_t i = 0; i < pDest->info.numOfCols; ++i) { - int32_t mapIndex = i; - // if (pIndexMap) { - // mapIndex = *(int32_t*)taosArrayGet(pIndexMap, i); - // } - SColumnInfoData* pCol2 = taosArrayGet(pDest->pDataBlock, i); - SColumnInfoData* pCol1 = taosArrayGet(pSrc->pDataBlock, mapIndex); + SColumnInfoData* pCol1 = taosArrayGet(pSrc->pDataBlock, i); capacity = pDest->info.capacity; colDataMergeCol(pCol2, pDest->info.rows, &capacity, pCol1, pSrc->info.rows); diff --git a/source/libs/command/src/explain.c b/source/libs/command/src/explain.c index 1acc9368c85fed790f43961b1bd7b23de00565c8..26a0f3bf6cf85bfe4d81a0ab5d8913d7e1767eeb 100644 --- a/source/libs/command/src/explain.c +++ b/source/libs/command/src/explain.c @@ -16,6 +16,7 @@ #include "commandInt.h" #include "plannodes.h" #include "query.h" +#include "tcommon.h" int32_t qExplainGenerateResNode(SPhysiNode *pNode, SExplainGroup *group, SExplainResNode **pRes); int32_t qExplainAppendGroupResRows(void *pCtx, int32_t groupId, int32_t level); @@ -637,13 +638,48 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen)); EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); } - EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT, pSortNode->pSortKeys->length); + + SDataBlockDescNode* pDescNode = pSortNode->node.pOutputDataBlockDesc; + EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT, nodesGetOutputNumFromSlotList(pDescNode->pSlots)); EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); - EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pSortNode->node.pOutputDataBlockDesc->totalRowSize); + EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pDescNode->totalRowSize); EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT); EXPLAIN_ROW_END(); QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level)); + if (EXPLAIN_MODE_ANALYZE == ctx->mode) { + // sort key + EXPLAIN_ROW_NEW(level, "Sort Key: "); + if (pResNode->pExecInfo) { + for (int32_t i = 0; i < LIST_LENGTH(pSortNode->pSortKeys); ++i) { + SOrderByExprNode *ptn = nodesListGetNode(pSortNode->pSortKeys, i); + EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr)); + } + } + + EXPLAIN_ROW_END(); + QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level)); + + // sort method + EXPLAIN_ROW_NEW(level, "Sort Method: "); + + int32_t nodeNum = taosArrayGetSize(pResNode->pExecInfo); + SExplainExecInfo *execInfo = taosArrayGet(pResNode->pExecInfo, 0); + SSortExecInfo * pExecInfo = (SSortExecInfo *)execInfo->verboseInfo; + EXPLAIN_ROW_APPEND("%s", pExecInfo->sortMethod == SORT_QSORT_T ? "quicksort" : "merge sort"); + if (pExecInfo->sortBuffer > 1024 * 1024) { + EXPLAIN_ROW_APPEND(" Buffers:%.2f Mb", pExecInfo->sortBuffer / (1024 * 1024.0)); + } else if (pExecInfo->sortBuffer > 1024) { + EXPLAIN_ROW_APPEND(" Buffers:%.2f Kb", pExecInfo->sortBuffer / (1024.0)); + } else { + EXPLAIN_ROW_APPEND(" Buffers:%d b", pExecInfo->sortBuffer); + } + + EXPLAIN_ROW_APPEND(" loops:%d", pExecInfo->loops); + EXPLAIN_ROW_END(); + QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level)); + } + if (verbose) { EXPLAIN_ROW_NEW(level + 1, EXPLAIN_OUTPUT_FORMAT); EXPLAIN_ROW_APPEND(EXPLAIN_COLUMNS_FORMAT, @@ -792,13 +828,8 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen)); EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); } -// EXPLAIN_ROW_APPEND(EXPLAIN_FUNCTIONS_FORMAT, pPartNode->length); -// EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pPartNode->node.pOutputDataBlockDesc->totalRowSize); -// if (pPartNode->pGroupKeys) { -// EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); -// EXPLAIN_ROW_APPEND(EXPLAIN_GROUPS_FORMAT, pPartNode->pGroupKeys->length); -// } + EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT); EXPLAIN_ROW_END(); QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level)); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 8ac320b9aa2323b9ec92b6062c99da5c2b2452c9..2dfbaaa4aac55d6d6b85f14c8880847dae0b5297 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -588,18 +588,14 @@ typedef struct SSortedMergeOperatorInfo { typedef struct SSortOperatorInfo { SOptrBasicInfo binfo; - uint32_t sortBufSize; // max buffer size for in-memory sort + uint32_t sortBufSize; // max buffer size for in-memory sort SArray* pSortInfo; SSortHandle* pSortHandle; SArray* pColMatchInfo; // for index map from table scan output int32_t bufPageSize; - // TODO extact struct - int64_t startTs; // sort start time - uint64_t sortElapsed; // sort elapsed time, time to flush to disk not included. - uint64_t totalSize; // total load bytes from remote - uint64_t totalRows; // total number of rows - uint64_t totalElapsed; // total elapsed time + int64_t startTs; // sort start time + uint64_t sortElapsed; // sort elapsed time, time to flush to disk not included. } SSortOperatorInfo; typedef struct STagFilterOperatorInfo { diff --git a/source/libs/executor/inc/tsort.h b/source/libs/executor/inc/tsort.h index d74628a72fb4723d1837a0547574da414253bef6..c8b1b3ee513bc508de5187c8d39ace4ae5e4b7f8 100644 --- a/source/libs/executor/inc/tsort.h +++ b/source/libs/executor/inc/tsort.h @@ -137,6 +137,14 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colId); */ SSDataBlock* tsortGetSortedDataBlock(const SSortHandle* pSortHandle); +/** + * return the sort execution information. + * + * @param pHandle + * @return + */ +SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle); + #ifdef __cplusplus } #endif diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 2215d0035d82bdd7bcc4eb405235e9901b39f279..ccd4b7c4cfb0fa9a1e435ddb6861a93f48d0f001 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -790,7 +790,7 @@ static SSDataBlock* getDataFromCatch(SStreamBlockScanInfo* pInfo) { SSDataBlock* pDB = createOneDataBlock(pInfo->pRes, false); blockDataFromBuf(pDB, buf); SSDataBlock* pSub = blockDataExtractBlock(pDB, pos->rowId, 1); - blockDataMerge(pInfo->pRes, pSub, NULL); + blockDataMerge(pInfo->pRes, pSub); blockDataDestroy(pDB); blockDataDestroy(pSub); } diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index 990dc0f20002134eebc0cbe15a0fc4d0e34e6dc8..8f5fa88070fde1625385fd6e691ccccf424c2094 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -2,6 +2,9 @@ #include "executorimpl.h" static SSDataBlock* doSort(SOperatorInfo* pOperator); +static int32_t doOpenSortOperator(SOperatorInfo* pOperator); +static int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len); + static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput); SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExprInfo* pExprInfo, int32_t numOfCols, @@ -35,7 +38,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pR pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doSort, NULL, NULL, destroyOrderOperatorInfo, NULL, NULL, NULL); + createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroyOrderOperatorInfo, NULL, NULL, getExplainExecInfo); int32_t code = appendDownstream(pOperator, &downstream, 1); return pOperator; @@ -121,20 +124,17 @@ void applyScalarFunction(SSDataBlock* pBlock, void* param) { } } -SSDataBlock* doSort(SOperatorInfo* pOperator) { - if (pOperator->status == OP_EXEC_DONE) { - return NULL; - } - - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; +int32_t doOpenSortOperator(SOperatorInfo* pOperator) { SSortOperatorInfo* pInfo = pOperator->info; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - if (pOperator->status == OP_RES_TO_RETURN) { - return getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pInfo->pColMatchInfo); + if (OPTR_IS_OPENED(pOperator)) { + return TSDB_CODE_SUCCESS; } -// pInfo->binfo.pRes is not equalled to the input datablock. -// int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize; + pInfo->startTs = taosGetTimestampUs(); + + // pInfo->binfo.pRes is not equalled to the input datablock. pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_SINGLESOURCE_SORT, -1, -1, NULL, pTaskInfo->id.str); @@ -146,12 +146,39 @@ SSDataBlock* doSort(SOperatorInfo* pOperator) { int32_t code = tsortOpen(pInfo->pSortHandle); taosMemoryFreeClear(ps); + if (code != TSDB_CODE_SUCCESS) { longjmp(pTaskInfo->env, terrno); } + pOperator->cost.openCost = (taosGetTimestampUs() - pInfo->startTs)/1000.0; pOperator->status = OP_RES_TO_RETURN; - return getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pInfo->pColMatchInfo); + + OPTR_SET_OPENED(pOperator); + return TSDB_CODE_SUCCESS; +} + +SSDataBlock* doSort(SOperatorInfo* pOperator) { + if (pOperator->status == OP_EXEC_DONE) { + return NULL; + } + + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + SSortOperatorInfo* pInfo = pOperator->info; + + int32_t code = pOperator->fpSet._openFn(pOperator); + if (code != TSDB_CODE_SUCCESS) { + longjmp(pTaskInfo->env, code); + } + + SSDataBlock* pBlock = getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pInfo->pColMatchInfo); + + if (pBlock != NULL) { + pOperator->resultInfo.totalRows += pBlock->info.rows; + } else { + doSetOperatorCompleted(pOperator); + } + return pBlock; } void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) { @@ -161,3 +188,15 @@ void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) { taosArrayDestroy(pInfo->pSortInfo); taosArrayDestroy(pInfo->pColMatchInfo); } + +int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) { + ASSERT(pOptr != NULL); + SSortExecInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortExecInfo)); + + SSortOperatorInfo *pOperatorInfo = (SSortOperatorInfo*)pOptr->info; + + *pInfo = tsortGetSortExecInfo(pOperatorInfo->pSortHandle); + *pOptrExplain = pInfo; + *len = sizeof(SSortExecInfo); + return TSDB_CODE_SUCCESS; +} diff --git a/source/libs/executor/src/tsort.c b/source/libs/executor/src/tsort.c index c826cb68bfaa5053d6c6cf09aa409f4518df9dfc..7581836d595b2a01e119ddbbdea24b7cd9cb6a74 100644 --- a/source/libs/executor/src/tsort.c +++ b/source/libs/executor/src/tsort.c @@ -31,20 +31,16 @@ struct STupleHandle { struct SSortHandle { int32_t type; - int32_t pageSize; int32_t numOfPages; SDiskbasedBuf *pBuf; SArray *pSortInfo; - SArray *pIndexMap; SArray *pOrderedSource; - _sort_fetch_block_fn_t fetchfp; - _sort_merge_compar_fn_t comparFn; - SMultiwayMergeTreeInfo *pMergeTree; - int64_t startTs; + int32_t loops; uint64_t sortElapsed; + int64_t startTs; uint64_t totalElapsed; int32_t sourceId; @@ -53,13 +49,15 @@ struct SSortHandle { int32_t numOfCompletedSources; bool opened; const char *idStr; - bool inMemSort; bool needAdjust; STupleHandle tupleHandle; - void *param; void (*beforeFp)(SSDataBlock* pBlock, void* param); + + _sort_fetch_block_fn_t fetchfp; + _sort_merge_compar_fn_t comparFn; + SMultiwayMergeTreeInfo *pMergeTree; }; static int32_t msortComparFn(const void *pLeft, const void *pRight, void *param); @@ -80,7 +78,7 @@ SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, SArray* pIndexMap, int32_t pSortHandle->pageSize = pageSize; pSortHandle->numOfPages = numOfPages; pSortHandle->pSortInfo = pSortInfo; - pSortHandle->pIndexMap = pIndexMap; + pSortHandle->loops = 0; if (pBlock != NULL) { pSortHandle->pDataBlock = createOneDataBlock(pBlock, false); @@ -415,6 +413,9 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) { int32_t numOfRows = blockDataGetCapacityInRow(pHandle->pDataBlock, pHandle->pageSize); blockDataEnsureCapacity(pHandle->pDataBlock, numOfRows); + // the initial pass + sortPass + final mergePass + pHandle->loops = sortPass + 2; + size_t numOfSorted = taosArrayGetSize(pHandle->pOrderedSource); for(int32_t t = 0; t < sortPass; ++t) { int64_t st = taosGetTimestampUs(); @@ -502,12 +503,13 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) { return 0; } -static int32_t createInitialSortedMultiSources(SSortHandle* pHandle) { +static int32_t createInitialSources(SSortHandle* pHandle) { size_t sortBufSize = pHandle->numOfPages * pHandle->pageSize; if (pHandle->type == SORT_SINGLESOURCE_SORT) { SSortSource* source = taosArrayGetP(pHandle->pOrderedSource, 0); taosArrayClear(pHandle->pOrderedSource); + while (1) { SSDataBlock* pBlock = pHandle->fetchfp(source->param); if (pBlock == NULL) { @@ -524,6 +526,7 @@ static int32_t createInitialSortedMultiSources(SSortHandle* pHandle) { } else { pHandle->pageSize = 4096; } + // todo!! pHandle->numOfPages = 1024; sortBufSize = pHandle->numOfPages * pHandle->pageSize; @@ -535,7 +538,7 @@ static int32_t createInitialSortedMultiSources(SSortHandle* pHandle) { } // todo relocate the columns - int32_t code = blockDataMerge(pHandle->pDataBlock, pBlock, pHandle->pIndexMap); + int32_t code = blockDataMerge(pHandle->pDataBlock, pBlock); if (code != 0) { return code; } @@ -569,6 +572,7 @@ static int32_t createInitialSortedMultiSources(SSortHandle* pHandle) { pHandle->cmpParam.numOfSources = 1; pHandle->inMemSort = true; + pHandle->loops = 1; pHandle->tupleHandle.rowIndex = -1; pHandle->tupleHandle.pBlock = pHandle->pDataBlock; return 0; @@ -592,7 +596,7 @@ int32_t tsortOpen(SSortHandle* pHandle) { pHandle->opened = true; - int32_t code = createInitialSortedMultiSources(pHandle); + int32_t code = createInitialSources(pHandle); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -692,3 +696,20 @@ void* tsortGetValue(STupleHandle* pVHandle, int32_t colIndex) { SColumnInfoData* pColInfo = TARRAY_GET_ELEM(pVHandle->pBlock->pDataBlock, colIndex); return colDataGetData(pColInfo, pVHandle->rowIndex); } + +SSortExecInfo tsortGetSortExecInfo(SSortHandle* pHandle) { + SSortExecInfo info = {0}; + + info.sortBuffer = pHandle->pageSize * pHandle->numOfPages; + info.sortMethod = pHandle->inMemSort? SORT_QSORT_T:SORT_SPILLED_MERGE_SORT_T; + info.loops = pHandle->loops; + + if (pHandle->pBuf != NULL) { + SDiskbasedBufStatis st = getDBufStatis(pHandle->pBuf); + info.writeBytes = st.flushBytes; + info.readBytes = st.loadBytes; + } + + return info; +} + diff --git a/source/util/src/tpagedbuf.c b/source/util/src/tpagedbuf.c index 00f123370747fcc29eddbb9ad053514134d3bc8f..101ac78e1847a1db244f7dfe867f94aeec0447d4 100644 --- a/source/util/src/tpagedbuf.c +++ b/source/util/src/tpagedbuf.c @@ -549,11 +549,16 @@ void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) { // print the statistics information { SDiskbasedBufStatis* ps = &pBuf->statis; - uDebug( - "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f " - "Kb\n", - ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, - ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages)); + if (ps->loadPages == 0) { + uDebug( + "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages)", + ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, ps->loadPages); + } else { + uDebug( + "Get/Release pages:%d/%d, flushToDisk:%.2f Kb (%d Pages), loadFromDisk:%.2f Kb (%d Pages), avgPageSize:%.2f Kb", + ps->getPages, ps->releasePages, ps->flushBytes / 1024.0f, ps->flushPages, ps->loadBytes / 1024.0f, + ps->loadPages, ps->loadBytes / (1024.0 * ps->loadPages)); + } } taosRemoveFile(pBuf->path);