feat: prepare for multiple group tsdb reads
This commit is contained in:
parent
fadc11b685
commit
c1f5f825b7
|
@ -195,6 +195,7 @@ struct SVnodeCfg {
|
|||
typedef struct {
|
||||
TSKEY lastKey;
|
||||
uint64_t uid;
|
||||
uint64_t groupId;
|
||||
} STableKeyInfo;
|
||||
|
||||
struct SMetaEntry {
|
||||
|
|
|
@ -2845,7 +2845,7 @@ int32_t tsdbGetAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) {
|
|||
break;
|
||||
}
|
||||
|
||||
STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
|
||||
STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id, .groupId = 0};
|
||||
taosArrayPush(list, &info);
|
||||
}
|
||||
|
||||
|
|
|
@ -834,14 +834,19 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi
|
|||
SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
|
||||
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
|
||||
|
||||
int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
|
||||
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
|
||||
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId,
|
||||
SNode* pTagCond);
|
||||
int32_t doCreateMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
|
||||
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
|
||||
uint64_t taskId, SNode* pTagCond);
|
||||
uint64_t taskId);
|
||||
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, SArray* dataReaders,
|
||||
SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
|
||||
|
||||
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
|
||||
|
||||
int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SArray* groupKey);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
|
|
|
@ -239,7 +239,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
|
|||
}
|
||||
|
||||
for (int i = 0; i < taosArrayGetSize(res); i++) {
|
||||
STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, .uid = *(uint64_t*)taosArrayGet(res, i)};
|
||||
STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, .uid = *(uint64_t*)taosArrayGet(res, i), .groupId = 0};
|
||||
taosArrayPush(pListInfo->pTableList, &info);
|
||||
}
|
||||
taosArrayDestroy(res);
|
||||
|
@ -247,7 +247,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
|
|||
code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList);
|
||||
}
|
||||
} else { // Create one table group.
|
||||
STableKeyInfo info = {.lastKey = 0, .uid = tableUid};
|
||||
STableKeyInfo info = {.lastKey = 0, .uid = tableUid, .groupId = 0};
|
||||
taosArrayPush(pListInfo->pTableList, &info);
|
||||
}
|
||||
|
||||
|
|
|
@ -3968,14 +3968,16 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
int32_t len = (int32_t)(pStart - (char*)keyBuf);
|
||||
uint64_t* groupId = taosHashGet(pTableListInfo->map, keyBuf, len);
|
||||
if (groupId) {
|
||||
taosHashPut(pTableListInfo->map, &(info->uid), sizeof(uint64_t), groupId, sizeof(uint64_t));
|
||||
} else {
|
||||
|
||||
uint64_t* pGroupId = taosHashGet(pTableListInfo->map, keyBuf, len);
|
||||
|
||||
if (!pGroupId) {
|
||||
uint64_t tmpId = calcGroupId(keyBuf, len);
|
||||
info->groupId = tmpId;
|
||||
taosHashPut(pTableListInfo->map, &(info->uid), sizeof(uint64_t), &tmpId, sizeof(uint64_t));
|
||||
} else {
|
||||
info->groupId = *pGroupId;
|
||||
}
|
||||
|
||||
metaReaderClear(&mr);
|
||||
|
@ -4023,11 +4025,11 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
|
|||
STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
|
||||
|
||||
SArray* dataReaders = taosArrayInit(8, POINTER_BYTES);
|
||||
createMultipleDataReaders(pTableScanNode, pHandle, pTableListInfo, dataReaders, queryId, taskId, pTagCond);
|
||||
createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId, pTagCond);
|
||||
doCreateMultipleDataReaders(pTableScanNode, pHandle, pTableListInfo, dataReaders, queryId, taskId);
|
||||
|
||||
extractTableSchemaVersion(pHandle, pTableScanNode->scan.uid, pTaskInfo);
|
||||
SArray* groupKeys = extractPartitionColInfo(pTableScanNode->pPartitionTags);
|
||||
generateGroupIdMap(pTableListInfo, pHandle, groupKeys); // todo for json
|
||||
taosArrayDestroy(groupKeys);
|
||||
|
||||
SOperatorInfo* pOperator = createTableMergeScanOperatorInfo(pTableScanNode, dataReaders, pHandle, pTaskInfo);
|
||||
STableScanInfo* pScanInfo = pOperator->info;
|
||||
pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
|
||||
|
|
|
@ -1872,7 +1872,6 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi
|
|||
SExprInfo* pExprInfo = createExprInfo(pPhyNode->pScanPseudoCols, NULL, &numOfExprs);
|
||||
SArray* colList = extractColMatchInfo(pPhyNode->pScanPseudoCols, pDescNode, &num, COL_MATCH_FROM_COL_ID);
|
||||
|
||||
|
||||
int32_t code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfExprs);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
goto _error;
|
||||
|
@ -1957,25 +1956,42 @@ typedef struct STableMergeScanInfo {
|
|||
|
||||
} STableMergeScanInfo;
|
||||
|
||||
int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
|
||||
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
|
||||
uint64_t taskId, SNode* pTagCond) {
|
||||
int32_t compareTableKeyInfoByGid(const void* p1, const void* p2) {
|
||||
const STableKeyInfo* info1 = p1;
|
||||
const STableKeyInfo* info2 = p2;
|
||||
return info1->groupId - info2->groupId;
|
||||
}
|
||||
|
||||
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
|
||||
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId,
|
||||
SNode* pTagCond) {
|
||||
int32_t code = getTableList(pHandle->meta, &pTableScanNode->scan, pTableListInfo, pTagCond);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
goto _error;
|
||||
return code;
|
||||
}
|
||||
|
||||
if (taosArrayGetSize(pTableListInfo->pTableList) == 0) {
|
||||
qDebug("no table qualified for query, TID:0x%" PRIx64 ", QID:0x%" PRIx64, taskId, queryId);
|
||||
goto _error;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
SArray* groupKeys = extractPartitionColInfo(pTableScanNode->pPartitionTags);
|
||||
generateGroupIdMap(pTableListInfo, pHandle, groupKeys); // todo for json
|
||||
if (groupKeys) {
|
||||
taosArraySort(pTableListInfo->pTableList, compareTableKeyInfoByGid);
|
||||
}
|
||||
taosArrayDestroy(groupKeys);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t doCreateMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
|
||||
STableListInfo* pTableListInfo, SArray* arrayReader, uint64_t queryId,
|
||||
uint64_t taskId) {
|
||||
|
||||
SQueryTableDataCond cond = {0};
|
||||
code = initQueryTableDataCond(&cond, pTableScanNode);
|
||||
int32_t code = initQueryTableDataCond(&cond, pTableScanNode);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
goto _error;
|
||||
}
|
||||
// TODO: free the sublist info and the table list in it
|
||||
for (int32_t i = 0; i < taosArrayGetSize(pTableListInfo->pTableList); ++i) {
|
||||
STableListInfo* subListInfo = taosMemoryCalloc(1, sizeof(subListInfo));
|
||||
subListInfo->pTableList = taosArrayInit(1, sizeof(STableKeyInfo));
|
||||
|
@ -1989,7 +2005,7 @@ int32_t createMultipleDataReaders(STableScanPhysiNode* pTableScanNode, SReadHand
|
|||
}
|
||||
cleanupQueryTableDataCond(&cond);
|
||||
|
||||
return 0;
|
||||
return TSDB_CODE_SUCCESS;
|
||||
|
||||
_error:
|
||||
return code;
|
||||
|
|
Loading…
Reference in New Issue