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

fix(tmq): add lock when check table list.

This commit is contained in:
Haojun Liao 2023-03-31 19:36:27 +08:00
parent 066a173d28
commit ae4234cce0
1 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -771,7 +771,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
// scan table one by one sequentially // scan table one by one sequentially
if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) { if (pInfo->scanMode == TABLE_SCAN__TABLE_ORDER) {
int32_t numOfTables = tableListGetSize(pTaskInfo->pTableInfoList); int32_t numOfTables = 0;//tableListGetSize(pTaskInfo->pTableInfoList);
STableKeyInfo tInfo = {0};
while (1) { while (1) {
SSDataBlock* result = doGroupedTableScan(pOperator); SSDataBlock* result = doGroupedTableScan(pOperator);
@ -781,14 +782,21 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
// if no data, switch to next table and continue scan // if no data, switch to next table and continue scan
pInfo->currentTable++; pInfo->currentTable++;
taosRLockLatch(&pTaskInfo->lock);
numOfTables = tableListGetSize(pTaskInfo->pTableInfoList);
if (pInfo->currentTable >= numOfTables) { if (pInfo->currentTable >= numOfTables) {
qDebug("all table checked in table list, total:%d, return NULL, %s", numOfTables, GET_TASKID(pTaskInfo)); qDebug("all table checked in table list, total:%d, return NULL, %s", numOfTables, GET_TASKID(pTaskInfo));
taosRUnLockLatch(&pTaskInfo->lock);
return NULL; return NULL;
} }
STableKeyInfo* pTableInfo = tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->currentTable); tInfo = *(STableKeyInfo*) tableListGetInfo(pTaskInfo->pTableInfoList, pInfo->currentTable);
tsdbSetTableList(pInfo->base.dataReader, pTableInfo, 1); taosRUnLockLatch(&pTaskInfo->lock);
qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d/%d %s", pTableInfo->uid, numOfTables,
tsdbSetTableList(pInfo->base.dataReader, &tInfo, 1);
qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d/%d %s", tInfo.uid, numOfTables,
pInfo->currentTable, numOfTables, GET_TASKID(pTaskInfo)); pInfo->currentTable, numOfTables, GET_TASKID(pTaskInfo));
tsdbReaderReset(pInfo->base.dataReader, &pInfo->base.cond); tsdbReaderReset(pInfo->base.dataReader, &pInfo->base.cond);