diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 5051bedf8de2221f883ac729a838ce1fce9f4f42..cd9f29978d644755f39adf2c44fdf6c5a9da57ff 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -251,10 +251,10 @@ typedef struct SLimitInfo { } SLimitInfo; typedef struct SExchangeInfo { - SArray* pSources; - SArray* pSourceDataInfo; - tsem_t ready; - void* pTransporter; + SArray* pSources; + SArray* pSourceDataInfo; + tsem_t ready; + void* pTransporter; // SArray, result block list, used to keep the multi-block that // passed by downstream operator SArray* pResultBlockList; diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 7e93232e0ce8c2ad85a9f72e6b971de078e3ec51..b7c3eed069e3d1c339a01fcd55e7c20c43a8f97c 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -1853,16 +1853,34 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn int64_t startTs = taosGetTimestampUs(); size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); + int32_t completed = 0; + for (int32_t k = 0; k < totalSources; ++k) { + SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k); + if (p->status == EX_SOURCE_DATA_EXHAUSTED) { + completed += 1; + } + } + + if (completed == totalSources) { + setAllSourcesCompleted(pOperator, startTs); + return; + } + while (1) { - int32_t completed = 0; +// printf("1\n"); + tsem_wait(&pExchangeInfo->ready); +// printf("2\n"); + for (int32_t i = 0; i < totalSources; ++i) { SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i); if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) { - completed += 1; +// printf("========:%d is completed\n", i); continue; } +// printf("index:%d - status:%d\n", i, pDataInfo->status); if (pDataInfo->status != EX_SOURCE_DATA_READY) { +// printf("-----------%d, status:%d, continue\n", i, pDataInfo->status); continue; } @@ -1874,22 +1892,31 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn SRetrieveTableRsp* pRsp = pDataInfo->pRsp; SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, i); + // todo SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo; if (pRsp->numOfRows == 0) { + pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; +// printf("%d completed, try next\n", i); + qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 ", completed:%d try next %d/%" PRIzu, GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows, - pExchangeInfo->loadInfo.totalRows, completed + 1, i + 1, totalSources); - pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; - completed += 1; + pExchangeInfo->loadInfo.totalRows, completed, i + 1, totalSources); taosMemoryFreeClear(pDataInfo->pRsp); - continue; + +// if (completed == totalSources) { +// return; +// } else { +// break; +// } + break; } SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp; int32_t index = 0; char* pStart = pRetrieveRsp->data; while (index++ < pRetrieveRsp->numOfBlocks) { + printf("results, numOfBLock: %d\n", pRetrieveRsp->numOfBlocks); SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false); code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart); if (code != 0) { @@ -1902,20 +1929,26 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator); +// int32_t completed = 0; if (pRsp->completed == 1) { + pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; + +// for (int32_t k = 0; k < totalSources; ++k) { +// SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k); +// if (p->status == EX_SOURCE_DATA_EXHAUSTED) { +// completed += 1; +// } +// } + qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 - " execId:%d" - " index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 - ", total:%.2f Kb," - " completed:%d try next %d/%" PRIzu, + " execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 + ", total:%.2f Kb, completed:%d try next %d/%" PRIzu, GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pRsp->numOfBlocks, pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0, - completed + 1, i + 1, totalSources); - completed += 1; - pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; + completed, i + 1, totalSources); } else { qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 - " execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb", + " execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb", GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRsp->numOfBlocks, pRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0); } @@ -1931,15 +1964,24 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn } } +// if (completed == totalSources) { +// setAllSourcesCompleted(pOperator, startTs); +// } + return; } + int32_t completed = 0; + for (int32_t k = 0; k < totalSources; ++k) { + SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k); + if (p->status == EX_SOURCE_DATA_EXHAUSTED) { + completed += 1; + } + } + if (completed == totalSources) { - setAllSourcesCompleted(pOperator, startTs); return; } - - sched_yield(); } _error: @@ -1970,6 +2012,7 @@ static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) { pOperator->cost.openCost = taosGetTimestampUs() - startTs; tsem_wait(&pExchangeInfo->ready); + tsem_post(&pExchangeInfo->ready); return TSDB_CODE_SUCCESS; } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index ee935837d0067466e63869afb35b122552fc63fc..6d130481233966c5bea5127f687d11dd57f13911 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -956,8 +956,12 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, pOperator->pTaskInfo = pTaskInfo; pInfo->metaCache.pTableMetaEntryCache = taosLRUCacheInit(1024*128, -1, .5); - taosLRUCacheSetStrictCapacity(pInfo->metaCache.pTableMetaEntryCache, false); + if (pInfo->metaCache.pTableMetaEntryCache == NULL) { + code = terrno; + goto _error; + } + taosLRUCacheSetStrictCapacity(pInfo->metaCache.pTableMetaEntryCache, false); pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, destroyTableScanOperatorInfo, getTableScannerExecInfo); diff --git a/source/util/src/tlrucache.c b/source/util/src/tlrucache.c index 0ea7258828daae1422efc868b5b7d666e4c65fe8..264883be4e71c99f1f3f3a1fbc872bf0c522728e 100644 --- a/source/util/src/tlrucache.c +++ b/source/util/src/tlrucache.c @@ -630,13 +630,16 @@ static int getDefaultCacheShardBits(size_t capacity) { SLRUCache *taosLRUCacheInit(size_t capacity, int numShardBits, double highPriPoolRatio) { if (numShardBits >= 20) { + terrno = TSDB_CODE_INVALID_PARA; return NULL; } if (highPriPoolRatio < 0.0 || highPriPoolRatio > 1.0) { + terrno = TSDB_CODE_INVALID_PARA; return NULL; } SLRUCache *cache = taosMemoryCalloc(1, sizeof(SLRUCache)); if (!cache) { + terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } @@ -648,7 +651,7 @@ SLRUCache *taosLRUCacheInit(size_t capacity, int numShardBits, double highPriPoo cache->shards = taosMemoryCalloc(numShards, sizeof(SLRUCacheShard)); if (!cache->shards) { taosMemoryFree(cache); - + terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } diff --git a/tests/script/tsim/parser/auto_create_tb.sim b/tests/script/tsim/parser/auto_create_tb.sim index 312964a1ab3d11e6b361eeb805870844c4b90313..7803557bdce930ba60b6c1ed34d849ba5b09df09 100644 --- a/tests/script/tsim/parser/auto_create_tb.sim +++ b/tests/script/tsim/parser/auto_create_tb.sim @@ -186,6 +186,7 @@ endi sql select t1, count(*), first(c9) from $stb partition by t1 order by t1 asc slimit 3 if $rows != 3 then + print expect 3, actual: $rows return -1 endi if $data(1)[1] != 1 then