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

add quick cmd

This commit is contained in:
yihaoDeng 2024-11-13 17:01:56 +08:00
parent eddc7918d3
commit 620c7627e9
3 changed files with 188 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/common/tmsg.h
View File

@ -1760,6 +1760,7 @@ typedef struct {
int64_t metaSize;
int64_t rawDataSize;
int64_t s3Size;
const char* dbname;
} SDbSizeStatisInfo;
typedef struct {

31
source/libs/executor/src/operator.c
View File

@ -29,10 +29,9 @@
#include "storageapi.h"
#include "tdatablock.h"
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
__optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn,
__optr_explain_fn_t explain, __optr_get_ext_fn_t nextExtFn, __optr_notify_fn_t notifyFn) {
__optr_close_fn_t closeFn, __optr_reqBuf_fn_t reqBufFn, __optr_explain_fn_t explain,
__optr_get_ext_fn_t nextExtFn, __optr_notify_fn_t notifyFn) {
SOperatorFpSet fpSet = {
._openFn = openFn,
.getNextFn = nextFn,
@ -288,7 +287,6 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand
int32_t type = nodeType(pPhyNode);
const char* idstr = GET_TASKID(pTaskInfo);
if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
SOperatorInfo* pOperator = NULL;
if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
@ -318,8 +316,8 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand
pTableListInfo->idInfo.suid = pTableScanNode->scan.suid;
pTableListInfo->idInfo.tableType = pTableScanNode->scan.tableType;
} else {
code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort, pHandle,
pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
if (code) {
pTaskInfo->code = code;
tableListDestroy(pTableListInfo);
@ -346,8 +344,8 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand
return terrno;
}
code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle,
pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle, pTableListInfo,
pTagCond, pTagIndexCond, pTaskInfo);
if (code) {
pTaskInfo->code = code;
tableListDestroy(pTableListInfo);
@ -653,7 +651,6 @@ int32_t createOperator(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHand
return code;
}
void destroyOperator(SOperatorInfo* pOperator) {
if (pOperator == NULL) {
return;
@ -756,12 +753,14 @@ int32_t mergeOperatorParams(SOperatorParam* pDst, SOperatorParam* pSrc) {
tSimpleHashSetFreeFp(pBatch->pBatchs, freeExchangeGetBasicOperatorParam);
int32_t code = tSimpleHashPut(pBatch->pBatchs, &pDExc->basic.vgId, sizeof(pDExc->basic.vgId), &pDExc->basic, sizeof(pDExc->basic));
int32_t code = tSimpleHashPut(pBatch->pBatchs, &pDExc->basic.vgId, sizeof(pDExc->basic.vgId), &pDExc->basic,
sizeof(pDExc->basic));
if (code) {
return code;
}
code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic, sizeof(pSExc->basic));
code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic,
sizeof(pSExc->basic));
if (code) {
return code;
}
@ -776,14 +775,16 @@ int32_t mergeOperatorParams(SOperatorParam* pDst, SOperatorParam* pSrc) {
}
} else {
SExchangeOperatorBatchParam* pBatch = pDst->value;
SExchangeOperatorBasicParam* pBasic = tSimpleHashGet(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId));
SExchangeOperatorBasicParam* pBasic =
tSimpleHashGet(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId));
if (pBasic) {
void* p = taosArrayAddAll(pBasic->uidList, pSExc->basic.uidList);
if (p == NULL) {
return terrno;
}
} else {
int32_t code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic, sizeof(pSExc->basic));
int32_t code = tSimpleHashPut(pBatch->pBatchs, &pSExc->basic.vgId, sizeof(pSExc->basic.vgId), &pSExc->basic,
sizeof(pSExc->basic));
if (code) {
return code;
}
@ -799,7 +800,6 @@ int32_t mergeOperatorParams(SOperatorParam* pDst, SOperatorParam* pSrc) {
return TSDB_CODE_SUCCESS;
}
int32_t setOperatorParams(struct SOperatorInfo* pOperator, SOperatorParam* pInput, SOperatorParamType type) {
SOperatorParam** ppParam = NULL;
SOperatorParam*** pppDownstramParam = NULL;
@ -867,7 +867,6 @@ int32_t setOperatorParams(struct SOperatorInfo* pOperator, SOperatorParam* pInpu
return TSDB_CODE_SUCCESS;
}
SSDataBlock* getNextBlockFromDownstream(struct SOperatorInfo* pOperator, int32_t idx) {
SSDataBlock* p = NULL;
int32_t code = getNextBlockFromDownstreamImpl(pOperator, idx, true, &p);
@ -931,5 +930,3 @@ int16_t getOperatorResultBlockId(struct SOperatorInfo* pOperator, int32_t idx) {
}
return pOperator->resultDataBlockId;
}

133
source/libs/executor/src/sysscanoperator.c
View File

@ -73,6 +73,7 @@ typedef struct SSysTableScanInfo {
SLimitInfo limitInfo;
int32_t tbnameSlotId;
STableListInfo* pTableListInfo;
SSDataBlock* pResBlock;
SReadHandle* pHandle;
SStorageAPI* pAPI;
} SSysTableScanInfo;
@ -2015,40 +2016,15 @@ _end:
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}
static SSDataBlock* sysTableBuildVgUsage(SOperatorInfo* pOperator) {
static int32_t buildVgDiskUsage(SOperatorInfo* pOperator, SDbSizeStatisInfo* pStaticsInfo) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStorageAPI* pAPI = &pTaskInfo->storageAPI;
SSysTableScanInfo* pInfo = pOperator->info;
SSDataBlock* p = NULL;
int32_t numOfRows = 0;
const char* db = NULL;
int32_t vgId = 0;
int32_t numOfCols = 0;
// the retrieve is executed on the mnode, so return tables that belongs to the information schema database.
if (pInfo->readHandle.mnd != NULL) {
// code = buildSysUsageInfo(pInfo, pOperator->resultInfo.capacity);
// QUERY_CHECK_CODE(code, lino, _end);
// code = doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
// QUERY_CHECK_CODE(code, lino, _end);
// pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
setOperatorCompleted(pOperator);
return NULL;
// return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}
if (pInfo->pCur == NULL) {
pInfo->pCur = pAPI->metaFn.openTableMetaCursor(pInfo->readHandle.vnode);
}
blockDataCleanup(pInfo->pRes);
pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &db, &vgId, NULL, NULL);
const char* db = NULL;
pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, &pStaticsInfo->dbname, &vgId, NULL, NULL);
SDbSizeStatisInfo staticsInfo = {0};
@ -2057,12 +2033,77 @@ static SSDataBlock* sysTableBuildVgUsage(SOperatorInfo* pOperator) {
code = pAPI->metaFn.getDBSize(pInfo->readHandle.vnode, &staticsInfo);
QUERY_CHECK_CODE(code, lino, _end);
code = vnodeEstimateRawDataSize(pOperator, &staticsInfo);
return vnodeEstimateRawDataSize(pOperator, &staticsInfo);
_end:
return code;
}
static SSDataBlock* sysTableBuildVgUsage(SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SStorageAPI* pAPI = &pTaskInfo->storageAPI;
SSysTableScanInfo* pInfo = pOperator->info;
SDbSizeStatisInfo staticsInfo = {0};
SSDataBlock* p = NULL;
int32_t numOfRows = 0;
const char* db = NULL;
int32_t numOfCols = 0;
// the retrieve is executed on the mnode, so return tables that belongs to the information schema database.
if (pInfo->readHandle.mnd != NULL) {
setOperatorCompleted(pOperator);
return NULL;
}
if (pInfo->pCur == NULL) {
pInfo->pCur = pAPI->metaFn.openTableMetaCursor(pInfo->readHandle.vnode);
}
SSDataBlock* pBlock = pInfo->pRes;
code = buildVgDiskUsage(pOperator, &staticsInfo);
QUERY_CHECK_CODE(code, lino, _end);
if (pInfo->showRewrite) {
SSDataBlock* pBlock = pInfo->pRes;
SDBBlockUsageInfo usageInfo = {0};
int32_t len = tSerializeBlockDbUsage(NULL, 0, &usageInfo);
usageInfo.numOfSttRows = 120;
usageInfo.numOfInmemRows = 120;
usageInfo.numOfBlocks = 120;
char* p = taosMemoryCalloc(1, len + VARSTR_HEADER_SIZE);
QUERY_CHECK_NULL(p, code, lino, _end, terrno);
int32_t tempRes = tSerializeBlockDbUsage(varDataVal(p), len, &usageInfo);
if (tempRes != len) {
taosMemoryFree(p);
QUERY_CHECK_CODE(code, lino, _end);
}
varDataSetLen(p, len);
int32_t slotId = 1;
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, 1);
QUERY_CHECK_NULL(pColInfo, code, lino, _end, terrno);
code = colDataSetVal(pColInfo, 0, p, false);
QUERY_CHECK_CODE(code, lino, _end);
taosMemoryFree(p);
if (slotId != 0) {
SColumnInfoData* p1 = taosArrayGet(pBlock->pDataBlock, 0);
QUERY_CHECK_NULL(p1, code, lino, _end, terrno);
int64_t v = 0;
// colDataSetInt64(p1, 0, &v);
}
pBlock->info.rows = 1;
pOperator->status = OP_EXEC_DONE;
pInfo->pRes->info.rows = pBlock->info.rows;
// code = relocateColumnData(pInfo->pRes, pInfo->matchInfo.pList, pBlock->pDataBlock, false);
QUERY_CHECK_CODE(code, lino, _end);
} else {
SName sn = {0};
char dbname[TSDB_DB_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
code = tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);
code = tNameFromString(&sn, staticsInfo.dbname, T_NAME_ACCT | T_NAME_DB);
QUERY_CHECK_CODE(code, lino, _end);
code = tNameGetDbName(&sn, varDataVal(dbname));
@ -2081,7 +2122,7 @@ static SSDataBlock* sysTableBuildVgUsage(SOperatorInfo* pOperator) {
QUERY_CHECK_CODE(code, lino, _end);
pColInfoData = taosArrayGet(p->pDataBlock, numOfCols++);
code = colDataSetVal(pColInfoData, numOfRows, (char*)&vgId, false);
code = colDataSetVal(pColInfoData, numOfRows, (char*)&staticsInfo.vgId, false);
QUERY_CHECK_CODE(code, lino, _end);
pColInfoData = taosArrayGet(p->pDataBlock, numOfCols++);
@ -2111,15 +2152,11 @@ static SSDataBlock* sysTableBuildVgUsage(SOperatorInfo* pOperator) {
pColInfoData = taosArrayGet(p->pDataBlock, numOfCols++);
code = colDataSetVal(pColInfoData, numOfRows, (char*)&staticsInfo.rawDataSize, false); // estimate_size
QUERY_CHECK_CODE(code, lino, _end);
numOfRows += 1;
pAPI->metaFn.closeTableMetaCursor(pInfo->pCur);
pInfo->pCur = NULL;
if (numOfRows > 0) {
pAPI->metaFn.closeTableMetaCursor(pInfo->pCur);
pInfo->pCur = NULL;
p->info.rows = numOfRows;
pInfo->pRes->info.rows = numOfRows;
@ -2128,15 +2165,15 @@ static SSDataBlock* sysTableBuildVgUsage(SOperatorInfo* pOperator) {
code = doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
QUERY_CHECK_CODE(code, lino, _end);
numOfRows = 0;
}
blockDataDestroy(p);
p = NULL;
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
setOperatorCompleted(pOperator);
}
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
@ -2209,7 +2246,11 @@ _end:
}
return NULL;
}
static SSDataBlock* sysTableScanUsageRewrite(SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
return NULL;
}
static SSDataBlock* sysTableScanUsage(SOperatorInfo* pOperator) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
@ -2331,7 +2372,7 @@ static int32_t doSysTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes)
if (dbName[0]) TAOS_UNUSED(tsnprintf(pInfo->req.db, sizeof(pInfo->req.db), "%d.%s", pInfo->accountId, dbName));
(void)sysTableIsCondOnOneTable(pInfo->pCondition, pInfo->req.filterTb);
}
bool filter = true;
SSDataBlock* pBlock = NULL;
if (strncasecmp(name, TSDB_INS_TABLE_TABLES, TSDB_TABLE_FNAME_LEN) == 0) {
pBlock = sysTableScanUserTables(pOperator);
@ -2343,12 +2384,15 @@ static int32_t doSysTableScanNext(SOperatorInfo* pOperator, SSDataBlock** ppRes)
IS_SYS_DBNAME(dbName)) {
pBlock = sysTableScanUserSTables(pOperator);
} else if (strncasecmp(name, TSDB_INS_TABLE_USAGE, TSDB_TABLE_FNAME_LEN) == 0) {
if (pInfo->showRewrite) {
filter = false;
}
pBlock = sysTableScanUsage(pOperator);
} else { // load the meta from mnode of the given epset
pBlock = sysTableScanFromMNode(pOperator, pInfo, name, pTaskInfo);
}
sysTableScanFillTbName(pOperator, pInfo, name, pBlock);
if (filter) sysTableScanFillTbName(pOperator, pInfo, name, pBlock);
if (pBlock != NULL) {
bool limitReached = applyLimitOffset(&pInfo->limitInfo, pBlock, pTaskInfo);
if (limitReached) {
@ -2524,6 +2568,7 @@ int32_t createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNo
SScanPhysiNode* pScanNode = &pScanPhyNode->scan;
SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc;
QUERY_CHECK_CODE(code, lino, _error);
int32_t num = 0;
code = extractColMatchInfo(pScanNode->pScanCols, pDescNode, &num, COL_MATCH_FROM_COL_ID, &pInfo->matchInfo);
@ -2540,6 +2585,7 @@ int32_t createSysTableScanOperatorInfo(void* readHandle, SSystemTableScanPhysiNo
pInfo->showRewrite = pScanPhyNode->showRewrite;
pInfo->pRes = createDataBlockFromDescNode(pDescNode);
QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);
pInfo->pCondition = pScanNode->node.pConditions;
code = filterInitFromNode(pScanNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
QUERY_CHECK_CODE(code, lino, _error);
@ -3280,8 +3326,13 @@ static int32_t vnodeEstimateDataSizeByUid(SOperatorInfo* pOperator, STableId* id
void* pList = tableListGetInfo(pTableListInfo, 0);
code = pReadHandle->api.tsdReader.tsdReaderOpen(pReadHandle->vnode, &cond, pList, num, NULL, (void**)&pInfo->pHandle,
pTaskInfo->id.str, NULL);
if (pInfo->showRewrite) {
code = pReadHandle->api.tsdReader.tsdReaderOpen(pReadHandle->vnode, &cond, pList, num, pInfo->pRes,
(void**)&pInfo->pHandle, pTaskInfo->id.str, NULL);
} else {
code = pReadHandle->api.tsdReader.tsdReaderOpen(pReadHandle->vnode, &cond, pList, num, NULL,
(void**)&pInfo->pHandle, pTaskInfo->id.str, NULL);
}
cleanupQueryTableDataCond(&cond);
QUERY_CHECK_CODE(code, line, _end);