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

feat: add merge table scan operator

This commit is contained in:
slzhou 2022-06-15 16:40:43 +08:00
parent e60425b4a4
commit 562ec27c7b
2 changed files with 29 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
source/libs/executor/src/executorimpl.c
View File

@ -4675,32 +4675,33 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
//
// tsdbReaderT pDataReader =
// doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond);
// if (pDataReader == NULL && terrno != 0) {
// return NULL;
// }
//
// int32_t code = extractTableSchemaVersion(pHandle, pTableScanNode->scan.uid, pTaskInfo);
// if (code) {
// tsdbCleanupReadHandle(pDataReader);
// return NULL;
// }
//
// SArray* groupKeys = extractPartitionColInfo(pTableScanNode->pPartitionKeys);
// code = generateGroupIdMap(pTableListInfo, pHandle, groupKeys); //todo for json
// taosArrayDestroy(groupKeys);
// if (code){
// tsdbCleanupReadHandle(pDataReader);
// return NULL;
// }
//
// SOperatorInfo* pOperator =
// createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo);
//
// STableScanInfo* pScanInfo = pOperator->info;
// pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
tsdbReaderT pDataReader =
doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond);
if (pDataReader == NULL && terrno != 0) {
return NULL;
}
int32_t code = extractTableSchemaVersion(pHandle, pTableScanNode->scan.uid, pTaskInfo);
if (code) {
tsdbCleanupReadHandle(pDataReader);
return NULL;
}
SArray* groupKeys = extractPartitionColInfo(pTableScanNode->pPartitionKeys);
code = generateGroupIdMap(pTableListInfo, pHandle, groupKeys); // todo for json
taosArrayDestroy(groupKeys);
if (code) {
tsdbCleanupReadHandle(pDataReader);
return NULL;
}
SOperatorInfo* pOperator = createTableScanOperatorInfo(pTableScanNode, pDataReader, pHandle, pTaskInfo);
STableScanInfo* pScanInfo = pOperator->info;
pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
return pOperator;
} else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
SArray* dataReaders = taosArrayInit(8, POINTER_BYTES);
createMultipleDataReaders(pTableScanNode, pHandle, pTableListInfo, dataReaders, queryId, taskId, pTagCond);

4
source/libs/planner/src/planPhysiCreater.c
View File

@ -470,8 +470,8 @@ static ENodeType getScanOperatorType(EScanType scanType) {
case SCAN_TYPE_STREAM:
return QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
case SCAN_TYPE_TABLE_MERGE:
return QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
// return QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
// return QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
return QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
default:
break;
}