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

fix: select * from t core dump

This commit is contained in:
shenglian zhou 2022-06-14 23:34:50 +08:00
parent 3a971651db
commit a38f70ba7c
3 changed files with 37 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
source/libs/executor/inc/executorimpl.h
View File

@ -909,6 +909,11 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
int32_t getTableList(void* metaHandle, int32_t tableType, uint64_t tableUid, STableListInfo* pListInfo, int32_t getTableList(void* metaHandle, int32_t tableType, uint64_t tableUid, STableListInfo* pListInfo,
SNode* pTagCond); SNode* pTagCond);
int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
uint64_t taskId, SNode* pTagCond);
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SArray* dataReaders,
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

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

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

3
source/libs/executor/src/scanoperator.c
View File

@ -1865,7 +1865,7 @@ _error:
static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo, static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo,
int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) { int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STableScanInfo* pInfo = pOperator->info; STableMergeScanInfo* pInfo = pOperator->info;
SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder; SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder;
@ -2154,7 +2154,6 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) { void destroyTableMergeScanOperatorInfo(void* param, int32_t numOfOutput) {
STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param; STableMergeScanInfo* pTableScanInfo = (STableMergeScanInfo*)param;
blockDataDestroy(pTableScanInfo->pResBlock);
clearupQueryTableDataCond(&pTableScanInfo->cond); clearupQueryTableDataCond(&pTableScanInfo->cond);
for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->dataReaders); ++i) { for (int32_t i = 0; i < taosArrayGetSize(pTableScanInfo->dataReaders); ++i) {