diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 24ca3e2eedd009225073e22f75f9efe2cc4c30be..0173dd1c6cac1ef0c2313d1c2e5281eeb88e9544 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -32,18 +32,9 @@ #include "ttypes.h" #include "vnode.h" -typedef struct SBlockDistInfo { - SSDataBlock* pResBlock; - STsdbReader* pHandle; - SReadHandle readHandle; - uint64_t uid; // table uid -} SBlockDistInfo; - #define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN) #define SWITCH_ORDER(n) (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC)) - - typedef struct STableMergeScanExecInfo { SFileBlockLoadRecorder blockRecorder; SSortExecInfo sortExecInfo; @@ -946,169 +937,6 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* return pOperator; } -static int32_t doGetTableRowSize(void* pMeta, uint64_t uid, int32_t* rowLen, const char* idstr) { - *rowLen = 0; - - SMetaReader mr = {0}; - metaReaderInit(&mr, pMeta, 0); - int32_t code = metaGetTableEntryByUid(&mr, uid); - if (code != TSDB_CODE_SUCCESS) { - qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", uid, tstrerror(terrno), idstr); - metaReaderClear(&mr); - return terrno; - } - - if (mr.me.type == TSDB_SUPER_TABLE) { - int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols; - for (int32_t i = 0; i < numOfCols; ++i) { - (*rowLen) += mr.me.stbEntry.schemaRow.pSchema[i].bytes; - } - } else if (mr.me.type == TSDB_CHILD_TABLE) { - uint64_t suid = mr.me.ctbEntry.suid; - tDecoderClear(&mr.coder); - code = metaGetTableEntryByUid(&mr, suid); - if (code != TSDB_CODE_SUCCESS) { - qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", suid, tstrerror(terrno), idstr); - metaReaderClear(&mr); - return terrno; - } - - int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols; - - for (int32_t i = 0; i < numOfCols; ++i) { - (*rowLen) += mr.me.stbEntry.schemaRow.pSchema[i].bytes; - } - } else if (mr.me.type == TSDB_NORMAL_TABLE) { - int32_t numOfCols = mr.me.ntbEntry.schemaRow.nCols; - for (int32_t i = 0; i < numOfCols; ++i) { - (*rowLen) += mr.me.ntbEntry.schemaRow.pSchema[i].bytes; - } - } - - metaReaderClear(&mr); - return TSDB_CODE_SUCCESS; -} - -static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) { - if (pOperator->status == OP_EXEC_DONE) { - return NULL; - } - - SBlockDistInfo* pBlockScanInfo = pOperator->info; - SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; - - STableBlockDistInfo blockDistInfo = {.minRows = INT_MAX, .maxRows = INT_MIN}; - int32_t code = doGetTableRowSize(pBlockScanInfo->readHandle.meta, pBlockScanInfo->uid, - (int32_t*)&blockDistInfo.rowSize, GET_TASKID(pTaskInfo)); - if (code != TSDB_CODE_SUCCESS) { - T_LONG_JMP(pTaskInfo->env, code); - } - - tsdbGetFileBlocksDistInfo(pBlockScanInfo->pHandle, &blockDistInfo); - blockDistInfo.numOfInmemRows = (int32_t)tsdbGetNumOfRowsInMemTable(pBlockScanInfo->pHandle); - - SSDataBlock* pBlock = pBlockScanInfo->pResBlock; - - int32_t slotId = pOperator->exprSupp.pExprInfo->base.resSchema.slotId; - SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, slotId); - - int32_t len = tSerializeBlockDistInfo(NULL, 0, &blockDistInfo); - char* p = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE); - tSerializeBlockDistInfo(varDataVal(p), len, &blockDistInfo); - varDataSetLen(p, len); - - colDataAppend(pColInfo, 0, p, false); - taosMemoryFree(p); - - pBlock->info.rows = 1; - pOperator->status = OP_EXEC_DONE; - return pBlock; -} - -static void destroyBlockDistScanOperatorInfo(void* param) { - SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param; - blockDataDestroy(pDistInfo->pResBlock); - tsdbReaderClose(pDistInfo->pHandle); - taosMemoryFreeClear(param); -} - -static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pCond) { - memset(pCond, 0, sizeof(SQueryTableDataCond)); - - pCond->order = TSDB_ORDER_ASC; - pCond->numOfCols = 1; - pCond->colList = taosMemoryCalloc(1, sizeof(SColumnInfo)); - if (pCond->colList == NULL) { - terrno = TSDB_CODE_QRY_OUT_OF_MEMORY; - return terrno; - } - - pCond->colList->colId = 1; - pCond->colList->type = TSDB_DATA_TYPE_TIMESTAMP; - pCond->colList->bytes = sizeof(TSKEY); - - pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX}; - pCond->suid = uid; - pCond->type = TIMEWINDOW_RANGE_CONTAINED; - pCond->startVersion = -1; - pCond->endVersion = -1; - - return TSDB_CODE_SUCCESS; -} - -SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode, - SExecTaskInfo* pTaskInfo) { - SBlockDistInfo* pInfo = taosMemoryCalloc(1, sizeof(SBlockDistInfo)); - SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); - if (pInfo == NULL || pOperator == NULL) { - pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; - goto _error; - } - - { - SQueryTableDataCond cond = {0}; - - int32_t code = initTableblockDistQueryCond(pBlockScanNode->suid, &cond); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } - - STableListInfo* pTableListInfo = pTaskInfo->pTableInfoList; - size_t num = tableListGetSize(pTableListInfo); - void* pList = tableListGetInfo(pTableListInfo, 0); - - code = tsdbReaderOpen(readHandle->vnode, &cond, pList, num, &pInfo->pHandle, pTaskInfo->id.str); - cleanupQueryTableDataCond(&cond); - if (code != 0) { - goto _error; - } - } - - pInfo->readHandle = *readHandle; - pInfo->uid = pBlockScanNode->suid; - - pInfo->pResBlock = createResDataBlock(pBlockScanNode->node.pOutputDataBlockDesc); - blockDataEnsureCapacity(pInfo->pResBlock, 1); - - int32_t numOfCols = 0; - SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols); - int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols); - if (code != TSDB_CODE_SUCCESS) { - goto _error; - } - - setOperatorInfo(pOperator, "DataBlockDistScanOperator", QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, false, - OP_NOT_OPENED, pInfo, pTaskInfo); - pOperator->fpSet = - createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, NULL); - return pOperator; - -_error: - taosMemoryFreeClear(pInfo); - taosMemoryFreeClear(pOperator); - return NULL; -} - static FORCE_INLINE void doClearBufferedBlocks(SStreamScanInfo* pInfo) { taosArrayClear(pInfo->pBlockLists); pInfo->validBlockIndex = 0; diff --git a/source/libs/executor/src/sysscanoperator.c b/source/libs/executor/src/sysscanoperator.c index f6a0c57b6681ffd3b2b8fd78fadb00e3c6b176a2..c5e1f2c2140ea74b84a0e0f72f132a0f3ab93f97 100644 --- a/source/libs/executor/src/sysscanoperator.c +++ b/source/libs/executor/src/sysscanoperator.c @@ -78,6 +78,13 @@ typedef struct MergeIndex { int len; } MergeIndex; +typedef struct SBlockDistInfo { + SSDataBlock* pResBlock; + STsdbReader* pHandle; + SReadHandle readHandle; + uint64_t uid; // table uid +} SBlockDistInfo; + static int32_t sysChkFilter__Comm(SNode* pNode); static int32_t sysChkFilter__DBName(SNode* pNode); static int32_t sysChkFilter__VgroupId(SNode* pNode); @@ -1771,4 +1778,167 @@ static int32_t optSysTabFilte(void* arg, SNode* cond, SArray* result) { return -1; } return -1; +} + +static int32_t doGetTableRowSize(void* pMeta, uint64_t uid, int32_t* rowLen, const char* idstr) { + *rowLen = 0; + + SMetaReader mr = {0}; + metaReaderInit(&mr, pMeta, 0); + int32_t code = metaGetTableEntryByUid(&mr, uid); + if (code != TSDB_CODE_SUCCESS) { + qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", uid, tstrerror(terrno), idstr); + metaReaderClear(&mr); + return terrno; + } + + if (mr.me.type == TSDB_SUPER_TABLE) { + int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols; + for (int32_t i = 0; i < numOfCols; ++i) { + (*rowLen) += mr.me.stbEntry.schemaRow.pSchema[i].bytes; + } + } else if (mr.me.type == TSDB_CHILD_TABLE) { + uint64_t suid = mr.me.ctbEntry.suid; + tDecoderClear(&mr.coder); + code = metaGetTableEntryByUid(&mr, suid); + if (code != TSDB_CODE_SUCCESS) { + qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", suid, tstrerror(terrno), idstr); + metaReaderClear(&mr); + return terrno; + } + + int32_t numOfCols = mr.me.stbEntry.schemaRow.nCols; + + for (int32_t i = 0; i < numOfCols; ++i) { + (*rowLen) += mr.me.stbEntry.schemaRow.pSchema[i].bytes; + } + } else if (mr.me.type == TSDB_NORMAL_TABLE) { + int32_t numOfCols = mr.me.ntbEntry.schemaRow.nCols; + for (int32_t i = 0; i < numOfCols; ++i) { + (*rowLen) += mr.me.ntbEntry.schemaRow.pSchema[i].bytes; + } + } + + metaReaderClear(&mr); + return TSDB_CODE_SUCCESS; +} + +static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) { + if (pOperator->status == OP_EXEC_DONE) { + return NULL; + } + + SBlockDistInfo* pBlockScanInfo = pOperator->info; + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + + STableBlockDistInfo blockDistInfo = {.minRows = INT_MAX, .maxRows = INT_MIN}; + int32_t code = doGetTableRowSize(pBlockScanInfo->readHandle.meta, pBlockScanInfo->uid, + (int32_t*)&blockDistInfo.rowSize, GET_TASKID(pTaskInfo)); + if (code != TSDB_CODE_SUCCESS) { + T_LONG_JMP(pTaskInfo->env, code); + } + + tsdbGetFileBlocksDistInfo(pBlockScanInfo->pHandle, &blockDistInfo); + blockDistInfo.numOfInmemRows = (int32_t)tsdbGetNumOfRowsInMemTable(pBlockScanInfo->pHandle); + + SSDataBlock* pBlock = pBlockScanInfo->pResBlock; + + int32_t slotId = pOperator->exprSupp.pExprInfo->base.resSchema.slotId; + SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, slotId); + + int32_t len = tSerializeBlockDistInfo(NULL, 0, &blockDistInfo); + char* p = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE); + tSerializeBlockDistInfo(varDataVal(p), len, &blockDistInfo); + varDataSetLen(p, len); + + colDataAppend(pColInfo, 0, p, false); + taosMemoryFree(p); + + pBlock->info.rows = 1; + pOperator->status = OP_EXEC_DONE; + return pBlock; +} + +static void destroyBlockDistScanOperatorInfo(void* param) { + SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param; + blockDataDestroy(pDistInfo->pResBlock); + tsdbReaderClose(pDistInfo->pHandle); + taosMemoryFreeClear(param); +} + +static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pCond) { + memset(pCond, 0, sizeof(SQueryTableDataCond)); + + pCond->order = TSDB_ORDER_ASC; + pCond->numOfCols = 1; + pCond->colList = taosMemoryCalloc(1, sizeof(SColumnInfo)); + if (pCond->colList == NULL) { + terrno = TSDB_CODE_QRY_OUT_OF_MEMORY; + return terrno; + } + + pCond->colList->colId = 1; + pCond->colList->type = TSDB_DATA_TYPE_TIMESTAMP; + pCond->colList->bytes = sizeof(TSKEY); + + pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX}; + pCond->suid = uid; + pCond->type = TIMEWINDOW_RANGE_CONTAINED; + pCond->startVersion = -1; + pCond->endVersion = -1; + + return TSDB_CODE_SUCCESS; +} + +SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDistScanPhysiNode* pBlockScanNode, + SExecTaskInfo* pTaskInfo) { + SBlockDistInfo* pInfo = taosMemoryCalloc(1, sizeof(SBlockDistInfo)); + SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); + if (pInfo == NULL || pOperator == NULL) { + pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; + goto _error; + } + + { + SQueryTableDataCond cond = {0}; + + int32_t code = initTableblockDistQueryCond(pBlockScanNode->suid, &cond); + if (code != TSDB_CODE_SUCCESS) { + goto _error; + } + + STableListInfo* pTableListInfo = pTaskInfo->pTableInfoList; + size_t num = tableListGetSize(pTableListInfo); + void* pList = tableListGetInfo(pTableListInfo, 0); + + code = tsdbReaderOpen(readHandle->vnode, &cond, pList, num, &pInfo->pHandle, pTaskInfo->id.str); + cleanupQueryTableDataCond(&cond); + if (code != 0) { + goto _error; + } + } + + pInfo->readHandle = *readHandle; + pInfo->uid = pBlockScanNode->suid; + + pInfo->pResBlock = createResDataBlock(pBlockScanNode->node.pOutputDataBlockDesc); + blockDataEnsureCapacity(pInfo->pResBlock, 1); + + int32_t numOfCols = 0; + SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols); + int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols); + if (code != TSDB_CODE_SUCCESS) { + goto _error; + } + + setOperatorInfo(pOperator, "DataBlockDistScanOperator", QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN, false, + OP_NOT_OPENED, pInfo, pTaskInfo); + pOperator->fpSet = + createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, NULL); + return pOperator; + + _error: + taosMemoryFreeClear(pInfo); + taosMemoryFreeClear(pOperator); + return NULL; } \ No newline at end of file