提交 99e90faa 编写于 作者: S Shengliang Guan

Merge remote-tracking branch 'origin/3.0' into fix/TD-20662

...@@ -692,7 +692,6 @@ TEST(testCase, insert_test) { ...@@ -692,7 +692,6 @@ TEST(testCase, insert_test) {
taos_free_result(pRes); taos_free_result(pRes);
taos_close(pConn); taos_close(pConn);
} }
#endif
TEST(testCase, projection_query_tables) { TEST(testCase, projection_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
...@@ -752,9 +751,6 @@ TEST(testCase, projection_query_tables) { ...@@ -752,9 +751,6 @@ TEST(testCase, projection_query_tables) {
taos_close(pConn); taos_close(pConn);
} }
#if 0
TEST(testCase, tsbs_perf_test) { TEST(testCase, tsbs_perf_test) {
TdThread qid[20] = {0}; TdThread qid[20] = {0};
...@@ -764,15 +760,16 @@ TEST(testCase, tsbs_perf_test) { ...@@ -764,15 +760,16 @@ TEST(testCase, tsbs_perf_test) {
getchar(); getchar();
} }
#endif
TEST(testCase, projection_query_stables) { TEST(testCase, projection_query_stables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr); ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use abc1"); TAOS_RES* pRes = taos_query(pConn, "use test");
taos_free_result(pRes); taos_free_result(pRes);
pRes = taos_query(pConn, "select ts from st1"); pRes = taos_query(pConn, "select * from meters limit 50000000");
if (taos_errno(pRes) != 0) { if (taos_errno(pRes) != 0) {
printf("failed to select from table, reason:%s\n", taos_errstr(pRes)); printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes); taos_free_result(pRes);
...@@ -785,14 +782,15 @@ TEST(testCase, projection_query_stables) { ...@@ -785,14 +782,15 @@ TEST(testCase, projection_query_stables) {
char str[512] = {0}; char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) { while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields); // int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str); // printf("%s\n", str);
} }
taos_free_result(pRes); taos_free_result(pRes);
taos_close(pConn); taos_close(pConn);
} }
#if 0
TEST(testCase, agg_query_tables) { TEST(testCase, agg_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr); ASSERT_NE(pConn, nullptr);
......
...@@ -138,7 +138,9 @@ static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, ...@@ -138,7 +138,9 @@ static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput,
toDataCacheEntry(pDispatcher, pInput, pBuf); toDataCacheEntry(pDispatcher, pInput, pBuf);
taosWriteQitem(pDispatcher->pDataBlocks, pBuf); taosWriteQitem(pDispatcher->pDataBlocks, pBuf);
*pContinue = (DS_BUF_LOW == updateStatus(pDispatcher) ? true : false);
int32_t status = updateStatus(pDispatcher);
*pContinue = (status == DS_BUF_LOW || status == DS_BUF_EMPTY);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
......
...@@ -75,7 +75,9 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn ...@@ -75,7 +75,9 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
} }
while (1) { while (1) {
qDebug("prepare wait for ready, %p, %s", pExchangeInfo, GET_TASKID(pTaskInfo));
tsem_wait(&pExchangeInfo->ready); tsem_wait(&pExchangeInfo->ready);
if (isTaskKilled(pTaskInfo)) { if (isTaskKilled(pTaskInfo)) {
longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED); longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
} }
...@@ -360,7 +362,7 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) { ...@@ -360,7 +362,7 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId); SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId);
if (pExchangeInfo == NULL) { if (pExchangeInfo == NULL) {
qWarn("failed to acquire exchange operator, since it may have been released"); qWarn("failed to acquire exchange operator, since it may have been released, %p", pExchangeInfo);
taosMemoryFree(pMsg->pData); taosMemoryFree(pMsg->pData);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -379,20 +381,23 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) { ...@@ -379,20 +381,23 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
pRsp->numOfBlocks = htonl(pRsp->numOfBlocks); pRsp->numOfBlocks = htonl(pRsp->numOfBlocks);
ASSERT(pRsp != NULL); ASSERT(pRsp != NULL);
qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d", pSourceDataInfo->taskId, index, pRsp->numOfBlocks, qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d, %p", pSourceDataInfo->taskId, index, pRsp->numOfBlocks,
pRsp->numOfRows); pRsp->numOfRows, pExchangeInfo);
} else { } else {
taosMemoryFree(pMsg->pData); taosMemoryFree(pMsg->pData);
pSourceDataInfo->code = code; pSourceDataInfo->code = code;
qDebug("%s fetch rsp received, index:%d, error:%s", pSourceDataInfo->taskId, index, tstrerror(code)); qDebug("%s fetch rsp received, index:%d, error:%s, %p", pSourceDataInfo->taskId, index, tstrerror(code), pExchangeInfo);
} }
pSourceDataInfo->status = EX_SOURCE_DATA_READY; pSourceDataInfo->status = EX_SOURCE_DATA_READY;
code = tsem_post(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
code = TAOS_SYSTEM_ERROR(code);
qError("failed to invoke post when fetch rsp is ready, code:%s, %p", tstrerror(code), pExchangeInfo);
}
tsem_post(&pExchangeInfo->ready);
taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId); taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
return code;
return TSDB_CODE_SUCCESS;
} }
int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) { int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) {
...@@ -444,9 +449,9 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas ...@@ -444,9 +449,9 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
return pTaskInfo->code; return pTaskInfo->code;
} }
qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %d/%" PRIzu, qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %p, %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId, GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId,
pSource->execId, sourceIndex, totalSources); pSource->execId, pExchangeInfo, sourceIndex, totalSources);
// send the fetch remote task result reques // send the fetch remote task result reques
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
......
...@@ -1849,12 +1849,12 @@ FETCH_NEXT_BLOCK: ...@@ -1849,12 +1849,12 @@ FETCH_NEXT_BLOCK:
prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex); prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes); copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA; pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
printDataBlock(pDelBlock, "stream scan delete data"); printDataBlock(pDelBlock, "stream scan delete data");
if (pInfo->tqReader) { if (pInfo->tqReader) {
blockDataDestroy(pDelBlock); blockDataDestroy(pDelBlock);
} }
if (pInfo->pDeleteDataRes->info.rows > 0) { if (pInfo->pDeleteDataRes->info.rows > 0) {
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
return pInfo->pDeleteDataRes; return pInfo->pDeleteDataRes;
} else { } else {
goto FETCH_NEXT_BLOCK; goto FETCH_NEXT_BLOCK;
......
...@@ -3364,22 +3364,23 @@ void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t num ...@@ -3364,22 +3364,23 @@ void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t num
} }
} }
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t waterMark, uint16_t type, void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex,
int32_t tsColIndex) { STimeWindowAggSupp* pTwSup) {
if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) { if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) {
SStreamPartitionOperatorInfo* pScanInfo = downstream->info; SStreamPartitionOperatorInfo* pScanInfo = downstream->info;
pScanInfo->tsColIndex = tsColIndex; pScanInfo->tsColIndex = tsColIndex;
} }
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
initDownStream(downstream->pDownstream[0], pAggSup, waterMark, type, tsColIndex); initDownStream(downstream->pDownstream[0], pAggSup, type, tsColIndex, pTwSup);
return; return;
} }
SStreamScanInfo* pScanInfo = downstream->info; SStreamScanInfo* pScanInfo = downstream->info;
pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type}; pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type};
if (!pScanInfo->pUpdateInfo) { if (!pScanInfo->pUpdateInfo) {
pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark); pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark);
} }
pScanInfo->twAggSup = *pTwSup;
} }
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, int64_t gap, int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, int64_t gap,
...@@ -4102,8 +4103,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh ...@@ -4102,8 +4103,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo, NULL); createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo, NULL);
if (downstream) { if (downstream) {
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType, initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
pInfo->primaryTsIndex);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
} }
return pOperator; return pOperator;
...@@ -4606,8 +4606,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys ...@@ -4606,8 +4606,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo, pTaskInfo); pInfo, pTaskInfo);
pOperator->fpSet = pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, NULL); createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, NULL);
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType, initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
pInfo->primaryTsIndex);
code = appendDownstream(pOperator, &downstream, 1); code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
......
...@@ -731,5 +731,65 @@ if $data32 != 1 then ...@@ -731,5 +731,65 @@ if $data32 != 1 then
goto loop9 goto loop9
endi endi
sql drop stream if exists streams5;
sql drop database if exists test5;
sql create database test5;
sql use test5;
sql create table tb (ts timestamp, a int);
sql insert into tb values (now + 1m , 1 );
sql create table b (c timestamp, d int, e int , f int, g double);
sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a);
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3;
print data00:$data00
print data01:$data01
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3;
print data00:$data00
print data01:$data01
sql insert into b values(1648791213001,1,2,2,2.0);
sql insert into b values(1648791213002,1,3,3,3.0);
sql insert into tb values(1648791213003,1);
sql select * from streamt;
print data00:$data00
print data01:$data01
sql delete from b where c >= 1648791213001 and c <= 1648791213002;
sql insert into b values(1648791223003,2,2,3,1.0); insert into b values(1648791223002,2,3,3,3.0);
sql insert into tb values (now + 1m , 1 );
sql select * from streamt;
print data00:$data00
print data01:$data01
sql insert into b(c,d) values (now + 6m , 6 );
sql delete from b where c >= 1648791213001 and c <= 1648791233005;;
$loop_count = 0
loop10:
sleep 200
sql select c2 from streamt;
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
if $rows != 1 then
print =====rows=$rows
goto loop10
endi
if $data00 != 2 then
print =====data00=$data00
goto loop10
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode1 -s stop -x SIGINT
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册