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

Merge branch '3.0' into fix/TS-4540

This commit is contained in:
WANG MINGMING 2024-03-07 23:31:18 +08:00 committed by GitHub
parent 95358f374e ae6b16d809
commit ad6da4ab54
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 134 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
source/libs/executor/src/streamcountwindowoperator.c
View File

@ -390,10 +390,12 @@ void* doStreamCountDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
buf = taosDecodeFixedI32(buf, &mapSize); buf = taosDecodeFixedI32(buf, &mapSize);
for (int32_t i = 0; i < mapSize; i++) { for (int32_t i = 0; i < mapSize; i++) {
SSessionKey key = {0}; SSessionKey key = {0};
SResultWindowInfo winfo = {0}; SCountWindowInfo curWin = {0};
buf = decodeSSessionKey(buf, &key); buf = decodeSSessionKey(buf, &key);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize); SBuffInfo buffInfo = {.rebuildWindow = false, .winBuffOp = NONE_WINDOW, .pCur = NULL};
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo)); setCountOutputBuf(&pInfo->streamAggSup, key.win.skey, key.groupId, &curWin, &buffInfo);
buf = decodeSResultWindowInfo(buf, &curWin.winInfo, pInfo->streamAggSup.resultRowSize);
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo, sizeof(SResultWindowInfo));
} }
// 2.twAggSup // 2.twAggSup
@ -694,6 +696,8 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->recvGetAll = false; pInfo->recvGetAll = false;
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT;
setOperatorInfo(pOperator, getStreamOpName(pOperator->operatorType), QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT, true,
OP_NOT_OPENED, pInfo, pTaskInfo);
// for stream // for stream
void* buff = NULL; void* buff = NULL;
int32_t len = 0; int32_t len = 0;
@ -704,8 +708,6 @@ SOperatorInfo* createStreamCountAggOperatorInfo(SOperatorInfo* downstream, SPhys
doStreamCountDecodeOpState(buff, len, pOperator, true); doStreamCountDecodeOpState(buff, len, pOperator, true);
taosMemoryFree(buff); taosMemoryFree(buff);
} }
setOperatorInfo(pOperator, getStreamOpName(pOperator->operatorType), QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT, true,
OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamCountAgg, NULL, destroyStreamCountAggOperatorInfo, pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamCountAgg, NULL, destroyStreamCountAggOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamCountReleaseState, streamCountReloadState); setOperatorStreamStateFn(pOperator, streamCountReleaseState, streamCountReloadState);

7
source/libs/executor/src/streameventwindowoperator.c
View File

@ -406,6 +406,7 @@ void* doStreamEventDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
if (!pInfo) { if (!pInfo) {
return buf; return buf;
} }
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
// 4.checksum // 4.checksum
int32_t dataLen = len - sizeof(uint32_t); int32_t dataLen = len - sizeof(uint32_t);
@ -423,6 +424,8 @@ void* doStreamEventDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
SSessionKey key = {0}; SSessionKey key = {0};
SResultWindowInfo winfo = {0}; SResultWindowInfo winfo = {0};
buf = decodeSSessionKey(buf, &key); buf = decodeSSessionKey(buf, &key);
pAggSup->stateStore.streamStateSessionAddIfNotExist(pAggSup->pState, &winfo.sessionWin, pAggSup->gap,
(void**)&winfo.pStatePos, &pAggSup->resultRowSize);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize); buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize);
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo)); tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo));
} }
@ -735,6 +738,8 @@ SOperatorInfo* createStreamEventAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->reCkBlock = false; pInfo->reCkBlock = false;
pInfo->recvGetAll = false; pInfo->recvGetAll = false;
setOperatorInfo(pOperator, "StreamEventAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
// for stream // for stream
void* buff = NULL; void* buff = NULL;
int32_t len = 0; int32_t len = 0;
@ -746,8 +751,6 @@ SOperatorInfo* createStreamEventAggOperatorInfo(SOperatorInfo* downstream, SPhys
taosMemoryFree(buff); taosMemoryFree(buff);
} }
setOperatorInfo(pOperator, "StreamEventAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamEventAgg, NULL, destroyStreamEventOperatorInfo, pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamEventAgg, NULL, destroyStreamEventOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamEventReleaseState, streamEventReloadState); setOperatorStreamStateFn(pOperator, streamEventReleaseState, streamEventReloadState);

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

@ -2533,7 +2533,6 @@ int32_t encodeSResultWindowInfo(void** buf, SResultWindowInfo* key, int32_t outL
void* decodeSResultWindowInfo(void* buf, SResultWindowInfo* key, int32_t outLen) { void* decodeSResultWindowInfo(void* buf, SResultWindowInfo* key, int32_t outLen) {
buf = taosDecodeFixedBool(buf, &key->isOutput); buf = taosDecodeFixedBool(buf, &key->isOutput);
key->pStatePos->pRowBuff = NULL;
buf = decodeSSessionKey(buf, &key->sessionWin); buf = decodeSSessionKey(buf, &key->sessionWin);
return buf; return buf;
} }
@ -2591,6 +2590,7 @@ void* doStreamSessionDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
if (!pInfo) { if (!pInfo) {
return buf; return buf;
} }
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
// 5.checksum // 5.checksum
if (isParent) { if (isParent) {
@ -2609,6 +2609,8 @@ void* doStreamSessionDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOpera
SSessionKey key = {0}; SSessionKey key = {0};
SResultWindowInfo winfo = {0}; SResultWindowInfo winfo = {0};
buf = decodeSSessionKey(buf, &key); buf = decodeSSessionKey(buf, &key);
pAggSup->stateStore.streamStateSessionAddIfNotExist(pAggSup->pState, &winfo.sessionWin, pAggSup->gap,
(void**)&winfo.pStatePos, &pAggSup->resultRowSize);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize); buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize);
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo)); tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo));
} }
@ -2992,6 +2994,8 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
pInfo->recvGetAll = false; pInfo->recvGetAll = false;
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
setOperatorInfo(pOperator, getStreamOpName(pOperator->operatorType), QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION, true,
OP_NOT_OPENED, pInfo, pTaskInfo);
// for stream // for stream
void* buff = NULL; void* buff = NULL;
int32_t len = 0; int32_t len = 0;
@ -3002,8 +3006,6 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
doStreamSessionDecodeOpState(buff, len, pOperator, true); doStreamSessionDecodeOpState(buff, len, pOperator, true);
taosMemoryFree(buff); taosMemoryFree(buff);
} }
setOperatorInfo(pOperator, getStreamOpName(pOperator->operatorType), QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION, true,
OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo, pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamSessionReleaseState, streamSessionReloadState); setOperatorStreamStateFn(pOperator, streamSessionReleaseState, streamSessionReloadState);
@ -3538,6 +3540,7 @@ void* doStreamStateDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
if (!pInfo) { if (!pInfo) {
return buf; return buf;
} }
SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
// 5.checksum // 5.checksum
if (isParent) { if (isParent) {
@ -3556,6 +3559,9 @@ void* doStreamStateDecodeOpState(void* buf, int32_t len, SOperatorInfo* pOperato
SSessionKey key = {0}; SSessionKey key = {0};
SResultWindowInfo winfo = {0}; SResultWindowInfo winfo = {0};
buf = decodeSSessionKey(buf, &key); buf = decodeSSessionKey(buf, &key);
pAggSup->stateStore.streamStateStateAddIfNotExist(pAggSup->pState, &winfo.sessionWin, NULL,
pAggSup->stateKeySize, compareStateKey,
(void**)&winfo.pStatePos, &pAggSup->resultRowSize);
buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize); buf = decodeSResultWindowInfo(buf, &winfo, pInfo->streamAggSup.resultRowSize);
tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo)); tSimpleHashPut(pInfo->streamAggSup.pResultRows, &key, sizeof(SSessionKey), &winfo, sizeof(SResultWindowInfo));
} }
@ -3873,6 +3879,8 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT); pInfo->pCheckpointRes = createSpecialDataBlock(STREAM_CHECKPOINT);
pInfo->recvGetAll = false; pInfo->recvGetAll = false;
setOperatorInfo(pOperator, "StreamStateAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
// for stream // for stream
void* buff = NULL; void* buff = NULL;
int32_t len = 0; int32_t len = 0;
@ -3884,8 +3892,6 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
taosMemoryFree(buff); taosMemoryFree(buff);
} }
setOperatorInfo(pOperator, "StreamStateAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE, true, OP_NOT_OPENED,
pInfo, pTaskInfo);
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
setOperatorStreamStateFn(pOperator, streamStateReleaseState, streamStateReloadState); setOperatorStreamStateFn(pOperator, streamStateReleaseState, streamStateReloadState);

6
source/libs/parser/src/parTranslater.c
View File

@ -4547,7 +4547,7 @@ static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = getQueryTimeRange(pCxt, pSelect->pWhere, &pSelect->timeRange); code = getQueryTimeRange(pCxt, pSelect->pWhere, &pSelect->timeRange);
} }
if (pSelect->pWhere != NULL) { if (pSelect->pWhere != NULL && pCxt->pParseCxt->topicQuery == false) {
setTableVgroupsFromEqualTbnameCond(pCxt, pSelect); setTableVgroupsFromEqualTbnameCond(pCxt, pSelect);
} }
return code; return code;
@ -8135,7 +8135,9 @@ static int32_t adjustTagsForCreateTable(STranslateContext* pCxt, SCreateStreamSt
SColumnDefNode* pDef = (SColumnDefNode*)pTagDef; SColumnDefNode* pDef = (SColumnDefNode*)pTagDef;
if (!dataTypeEqual(&pDef->dataType, &((SExprNode*)pTagExpr)->resType)) { if (!dataTypeEqual(&pDef->dataType, &((SExprNode*)pTagExpr)->resType)) {
SNode* pFunc = NULL; SNode* pFunc = NULL;
int32_t code = createCastFunc(pCxt, pTagExpr, pDef->dataType, &pFunc); SDataType defType = pDef->dataType;
defType.bytes = calcTypeBytes(defType);
int32_t code = createCastFunc(pCxt, pTagExpr, defType, &pFunc);
if (TSDB_CODE_SUCCESS != code) { if (TSDB_CODE_SUCCESS != code) {
return code; return code;
} }

4
source/libs/scheduler/src/schRemote.c
View File

@ -940,6 +940,10 @@ int32_t schAsyncSendMsg(SSchJob *pJob, SSchTask *pTask, SSchTrans *trans, SQuery
SCH_ERR_JRET(schGenerateCallBackInfo(pJob, pTask, msg, msgSize, msgType, trans, isHb, &pMsgSendInfo)); SCH_ERR_JRET(schGenerateCallBackInfo(pJob, pTask, msg, msgSize, msgType, trans, isHb, &pMsgSendInfo));
SCH_ERR_JRET(schUpdateSendTargetInfo(pMsgSendInfo, addr, pTask)); SCH_ERR_JRET(schUpdateSendTargetInfo(pMsgSendInfo, addr, pTask));
if (isHb && persistHandle && trans->pHandle == 0) {
trans->pHandle = rpcAllocHandle();
}
if (pJob && pTask) { if (pJob && pTask) {
SCH_TASK_DLOG("start to send %s msg to node[%d,%s,%d], pTrans:%p, pHandle:%p", TMSG_INFO(msgType), addr->nodeId, SCH_TASK_DLOG("start to send %s msg to node[%d,%s,%d], pTrans:%p, pHandle:%p", TMSG_INFO(msgType), addr->nodeId,
epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, trans->pTrans, trans->pHandle); epSet->eps[epSet->inUse].fqdn, epSet->eps[epSet->inUse].port, trans->pTrans, trans->pHandle);

2
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -906,6 +906,7 @@ int32_t chkpMayDelObsolete(void* arg, int64_t chkpId, char* path) {
return 0; return 0;
} }
#ifdef BUILD_NO_CALL
static int32_t chkpIdComp(const void* a, const void* b) { static int32_t chkpIdComp(const void* a, const void* b) {
int64_t x = *(int64_t*)a; int64_t x = *(int64_t*)a;
int64_t y = *(int64_t*)b; int64_t y = *(int64_t*)b;
@ -964,6 +965,7 @@ int32_t streamBackendLoadCheckpointInfo(void* arg) {
taosMemoryFree(chkpPath); taosMemoryFree(chkpPath);
return 0; return 0;
} }
#endif
#ifdef BUILD_NO_CALL #ifdef BUILD_NO_CALL
int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*** ppHandle, SArray* refs) { int32_t chkpGetAllDbCfHandle(SStreamMeta* pMeta, rocksdb_column_family_handle_t*** ppHandle, SArray* refs) {

56
source/libs/stream/test/backendTest.cpp
View File

@ -29,7 +29,7 @@ class BackendEnv : public ::testing::Test {
void *backendCreate() { void *backendCreate() {
const char *streamPath = "/tmp"; const char *streamPath = "/tmp";
void *p = NULL; void * p = NULL;
// char *absPath = NULL; // char *absPath = NULL;
// // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2); // // SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(streamPath, -1, 2);
@ -52,7 +52,7 @@ SStreamState *stateCreate(const char *path) {
} }
void *backendOpen() { void *backendOpen() {
streamMetaInit(); streamMetaInit();
const char *path = "/tmp/backend"; const char * path = "/tmp/backend";
SStreamState *p = stateCreate(path); SStreamState *p = stateCreate(path);
ASSERT(p != NULL); ASSERT(p != NULL);
@ -79,7 +79,7 @@ void *backendOpen() {
const char *val = "value data"; const char *val = "value data";
int32_t len = 0; int32_t len = 0;
char *newVal = NULL; char * newVal = NULL;
streamStateGet_rocksdb(p, &key, (void **)&newVal, &len); streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
ASSERT(len == strlen(val)); ASSERT(len == strlen(val));
} }
@ -100,7 +100,7 @@ void *backendOpen() {
const char *val = "value data"; const char *val = "value data";
int32_t len = 0; int32_t len = 0;
char *newVal = NULL; char * newVal = NULL;
int32_t code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len); int32_t code = streamStateGet_rocksdb(p, &key, (void **)&newVal, &len);
ASSERT(code != 0); ASSERT(code != 0);
} }
@ -130,7 +130,7 @@ void *backendOpen() {
winkey.groupId = 0; winkey.groupId = 0;
winkey.ts = tsArray[0]; winkey.ts = tsArray[0];
char *val = NULL; char * val = NULL;
int32_t len = 0; int32_t len = 0;
pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey); pCurr = streamStateSeekKeyNext_rocksdb(p, &winkey);
@ -157,7 +157,7 @@ void *backendOpen() {
key.ts = tsArray[i]; key.ts = tsArray[i];
key.exprIdx = i; key.exprIdx = i;
char *val = NULL; char * val = NULL;
int32_t len = 0; int32_t len = 0;
streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len); streamStateFuncGet_rocksdb(p, &key, (void **)&val, &len);
ASSERT(len == strlen("Value")); ASSERT(len == strlen("Value"));
@ -168,7 +168,7 @@ void *backendOpen() {
key.ts = tsArray[i]; key.ts = tsArray[i];
key.exprIdx = i; key.exprIdx = i;
char *val = NULL; char * val = NULL;
int32_t len = 0; int32_t len = 0;
streamStateFuncDel_rocksdb(p, &key); streamStateFuncDel_rocksdb(p, &key);
} }
@ -213,7 +213,7 @@ void *backendOpen() {
{ {
SSessionKey key; SSessionKey key;
memset(&key, 0, sizeof(key)); memset(&key, 0, sizeof(key));
char *val = NULL; char * val = NULL;
int32_t vlen = 0; int32_t vlen = 0;
code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen); code = streamStateSessionGetKVByCur_rocksdb(pCurr, &key, (void **)&val, &vlen);
ASSERT(code == 0); ASSERT(code == 0);
@ -260,7 +260,7 @@ void *backendOpen() {
SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]}; SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
key.groupId = (uint64_t)(i); key.groupId = (uint64_t)(i);
key.ts = tsArray[i]; key.ts = tsArray[i];
char *val = NULL; char * val = NULL;
int32_t vlen = 0; int32_t vlen = 0;
ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0); ASSERT(streamStateFillGet_rocksdb(p, &key, (void **)&val, &vlen) == 0);
taosMemoryFreeClear(val); taosMemoryFreeClear(val);
@ -272,7 +272,7 @@ void *backendOpen() {
SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key); SStreamStateCur *pCurr = streamStateFillGetCur_rocksdb(p, &key);
ASSERT(pCurr != NULL); ASSERT(pCurr != NULL);
char *val = NULL; char * val = NULL;
int32_t vlen = 0; int32_t vlen = 0;
ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen)); ASSERT(0 == streamStateFillGetKVByCur_rocksdb(pCurr, &key, (const void **)&val, &vlen));
ASSERT(vlen == strlen("Value")); ASSERT(vlen == strlen("Value"));
@ -296,7 +296,7 @@ void *backendOpen() {
SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]}; SWinKey key = {0}; // {.groupId = (uint64_t)(i), .ts = tsArray[i]};
key.groupId = (uint64_t)(i); key.groupId = (uint64_t)(i);
key.ts = tsArray[i]; key.ts = tsArray[i];
char *val = NULL; char * val = NULL;
int32_t vlen = 0; int32_t vlen = 0;
ASSERT(streamStateFillDel_rocksdb(p, &key) == 0); ASSERT(streamStateFillDel_rocksdb(p, &key) == 0);
taosMemoryFreeClear(val); taosMemoryFreeClear(val);
@ -338,7 +338,7 @@ void *backendOpen() {
char key[128] = {0}; char key[128] = {0};
sprintf(key, "tbname_%d", i); sprintf(key, "tbname_%d", i);
char *val = NULL; char * val = NULL;
int32_t len = 0; int32_t len = 0;
code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len); code = streamDefaultGet_rocksdb(p, key, (void **)&val, &len);
ASSERT(code == 0); ASSERT(code == 0);
@ -354,7 +354,7 @@ TEST_F(BackendEnv, checkOpen) {
SStreamState *p = (SStreamState *)backendOpen(); SStreamState *p = (SStreamState *)backendOpen();
int64_t tsStart = taosGetTimestampMs(); int64_t tsStart = taosGetTimestampMs();
{ {
void *pBatch = streamStateCreateBatch(); void * pBatch = streamStateCreateBatch();
int32_t size = 0; int32_t size = 0;
for (int i = 0; i < size; i++) { for (int i = 0; i < size; i++) {
char key[128] = {0}; char key[128] = {0};
@ -368,7 +368,7 @@ TEST_F(BackendEnv, checkOpen) {
streamStateDestroyBatch(pBatch); streamStateDestroyBatch(pBatch);
} }
{ {
void *pBatch = streamStateCreateBatch(); void * pBatch = streamStateCreateBatch();
int32_t size = 0; int32_t size = 0;
char valBuf[256] = {0}; char valBuf[256] = {0};
for (int i = 0; i < size; i++) { for (int i = 0; i < size; i++) {
@ -385,7 +385,7 @@ TEST_F(BackendEnv, checkOpen) {
// do checkpoint 2 // do checkpoint 2
taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2); taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 2);
{ {
void *pBatch = streamStateCreateBatch(); void * pBatch = streamStateCreateBatch();
int32_t size = 0; int32_t size = 0;
char valBuf[256] = {0}; char valBuf[256] = {0};
for (int i = 0; i < size; i++) { for (int i = 0; i < size; i++) {
@ -407,12 +407,22 @@ TEST_F(BackendEnv, checkOpen) {
// taosMkDir(dump); // taosMkDir(dump);
taosMulMkDir(dump); taosMulMkDir(dump);
SBkdMgt *mgt = bkdMgtCreate((char *)path); SBkdMgt *mgt = bkdMgtCreate((char *)path);
SArray *result = taosArrayInit(4, sizeof(void *)); SArray * result = taosArrayInit(4, sizeof(void *));
bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump); bkdMgtGetDelta(mgt, p->pTdbState->idstr, 3, result, (char *)dump);
taskDbDoCheckpoint(p->pTdbState->pOwner->pBackend, 4);
taosArrayClear(result);
bkdMgtGetDelta(mgt, p->pTdbState->idstr, 4, result, (char *)dump);
bkdMgtDestroy(mgt); bkdMgtDestroy(mgt);
streamStateClose((SStreamState *)p, true); streamStateClose((SStreamState *)p, true);
// {
// taosRemoveDir("/tmp/backend");
// const char * path = "/tmp/backend";
// SStreamState *p = stateCreate(path);
// }
taosRemoveDir(path); taosRemoveDir(path);
// streamStateClose((SStreamState *)p, true);
} }
TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; } TEST_F(BackendEnv, backendChkp) { const char *path = "/tmp"; }
@ -430,6 +440,20 @@ TEST_F(BackendEnv, backendUtil) {
ASSERT_EQ(nextPow2((uint32_t)(kvDict[i].k)), kvDict[i].v); ASSERT_EQ(nextPow2((uint32_t)(kvDict[i].k)), kvDict[i].v);
} }
} }
TEST_F(BackendEnv, oldBackendInit) {
const char *path = "/tmp/backend1";
taosMulMkDir(path);
{
SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(path, 10, 10);
streamBackendCleanup((void *)p);
}
{
SBackendWrapper *p = (SBackendWrapper *)streamBackendInit(path, 10, 10);
streamBackendCleanup((void *)p);
}
taosRemoveDir(path);
}
int main(int argc, char **argv) { int main(int argc, char **argv) {
testing::InitGoogleTest(&argc, argv); testing::InitGoogleTest(&argc, argv);

19
source/libs/transport/src/transCli.c
View File

@ -584,8 +584,8 @@ void* destroyConnPool(SCliThrd* pThrd) {
static SCliConn* getConnFromPool(SCliThrd* pThrd, char* key, bool* exceed) { static SCliConn* getConnFromPool(SCliThrd* pThrd, char* key, bool* exceed) {
void* pool = pThrd->pool; void* pool = pThrd->pool;
SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key) + 1);
STrans* pTranInst = pThrd->pTransInst; STrans* pTranInst = pThrd->pTransInst;
SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key) + 1);
if (plist == NULL) { if (plist == NULL) {
SConnList list = {0}; SConnList list = {0};
taosHashPut((SHashObj*)pool, key, strlen(key) + 1, (void*)&list, sizeof(list)); taosHashPut((SHashObj*)pool, key, strlen(key) + 1, (void*)&list, sizeof(list));
@ -867,17 +867,18 @@ static void cliDestroyConn(SCliConn* conn, bool clear) {
QUEUE_INIT(&conn->q); QUEUE_INIT(&conn->q);
conn->broken = true; conn->broken = true;
if (conn->list == NULL) {
conn->list = taosHashGet((SHashObj*)pThrd->pool, conn->dstAddr, strlen(conn->dstAddr));
}
if (conn->list != NULL) { if (conn->list) {
SConnList* connList = conn->list; SConnList* list = conn->list;
connList->list->numOfConn--; list->list->numOfConn--;
connList->size--; if (conn->status == ConnInPool) {
} else { list->size--;
if (pThrd->pool) {
SConnList* connList = taosHashGet((SHashObj*)pThrd->pool, conn->dstAddr, strlen(conn->dstAddr) + 1);
if (connList != NULL) connList->list->numOfConn--;
} }
} }
conn->list = NULL; conn->list = NULL;
pThrd->newConnCount--; pThrd->newConnCount--;

11
source/libs/transport/src/transSvr.c
View File

@ -159,7 +159,7 @@ static void uvStartSendResp(SSvrMsg* msg);
static void uvNotifyLinkBrokenToApp(SSvrConn* conn); static void uvNotifyLinkBrokenToApp(SSvrConn* conn);
static FORCE_INLINE void destroySmsg(SSvrMsg* smsg); static FORCE_INLINE void destroySmsg(SSvrMsg* smsg);
static FORCE_INLINE SSvrConn* createConn(void* hThrd); static FORCE_INLINE SSvrConn* createConn(void* hThrd);
static FORCE_INLINE void destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/); static FORCE_INLINE void destroyConn(SSvrConn* conn, bool clear /*clear handle or not*/);
static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn); static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn);
@ -527,6 +527,10 @@ void uvOnSendCb(uv_write_t* req, int status) {
if (!transQueueEmpty(&conn->srvMsgs)) { if (!transQueueEmpty(&conn->srvMsgs)) {
msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0); msg = (SSvrMsg*)transQueueGet(&conn->srvMsgs, 0);
if (msg->type == Register && conn->status == ConnAcquire) { if (msg->type == Register && conn->status == ConnAcquire) {
if (conn->regArg.init) {
transFreeMsg(conn->regArg.msg.pCont);
conn->regArg.init = 0;
}
conn->regArg.notifyCount = 0; conn->regArg.notifyCount = 0;
conn->regArg.init = 1; conn->regArg.init = 1;
conn->regArg.msg = msg->msg; conn->regArg.msg = msg->msg;
@ -1350,6 +1354,11 @@ void uvHandleRegister(SSvrMsg* msg, SWorkThrd* thrd) {
return; return;
} }
transQueuePop(&conn->srvMsgs); transQueuePop(&conn->srvMsgs);
if (conn->regArg.init) {
transFreeMsg(conn->regArg.msg.pCont);
conn->regArg.init = 0;
}
conn->regArg.notifyCount = 0; conn->regArg.notifyCount = 0;
conn->regArg.init = 1; conn->regArg.init = 1;
conn->regArg.msg = msg->msg; conn->regArg.msg = msg->msg;

12
tests/script/tsim/stream/state0.sim
View File

@ -21,6 +21,8 @@ print create stream streams1 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0
sql create stream streams1 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); sql create stream streams1 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a);
sleep 1000
sql insert into t1 values(1648791213000,1,2,3,1.0,1); sql insert into t1 values(1648791213000,1,2,3,1.0,1);
sql insert into t1 values(1648791213000,1,2,3,1.0,2); sql insert into t1 values(1648791213000,1,2,3,1.0,2);
$loop_count = 0 $loop_count = 0
@ -457,6 +459,8 @@ print create stream streams2 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0
sql create stream streams2 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); sql create stream streams2 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a);
sleep 1000
sql insert into t1 values(1648791212000,2,2,3,1.0,1); sql insert into t1 values(1648791212000,2,2,3,1.0,1);
sql insert into t1 values(1648791213000,1,2,3,1.0,1); sql insert into t1 values(1648791213000,1,2,3,1.0,1);
sql insert into t1 values(1648791213000,1,2,4,1.0,2); sql insert into t1 values(1648791213000,1,2,4,1.0,2);
@ -504,6 +508,9 @@ sql create table t1(ts timestamp, a int, b int , c int, d double, id int);
print create stream streams3 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a); print create stream streams3 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a);
sql create stream streams3 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a); sql create stream streams3 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a);
sleep 1000
sql insert into t1 values(1648791212000,1,2,3,1.0,1); sql insert into t1 values(1648791212000,1,2,3,1.0,1);
sql insert into t1 values(1648791213000,2,2,3,1.0,1); sql insert into t1 values(1648791213000,2,2,3,1.0,1);
sql insert into t1 values(1648791214000,3,2,4,1.0,2); sql insert into t1 values(1648791214000,3,2,4,1.0,2);
@ -557,6 +564,8 @@ print create stream if not exists streams4 trigger window_close IGNORE EXPIRED 0
sql create stream if not exists streams4 trigger window_close IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt4 as select _wstart AS startts, min(c1),count(c1) from t1 state_window(c1); sql create stream if not exists streams4 trigger window_close IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt4 as select _wstart AS startts, min(c1),count(c1) from t1 state_window(c1);
sleep 1000
sql insert into t1 (ts, c1) values (1668073288209, 11); sql insert into t1 (ts, c1) values (1668073288209, 11);
sql insert into t1 (ts, c1) values (1668073288210, 11); sql insert into t1 (ts, c1) values (1668073288210, 11);
sql insert into t1 (ts, c1) values (1668073288211, 11); sql insert into t1 (ts, c1) values (1668073288211, 11);
@ -745,6 +754,9 @@ sql create table b (c timestamp, d int, e int , f int, g double);
print create stream streams0 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a); print create stream streams0 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a);
sql create stream streams0 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a); sql create stream streams0 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a);
sleep 1000
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL); sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3; sql select * from streamt order by c1, c2, c3;

30
tests/system-test/7-tmq/tmq_taosx.py
View File

@ -366,6 +366,7 @@ class TDTestCase:
tdSql.execute(f"insert into `test`.b1 using `test`.`b`(key) tags('1') (time, task_id) values ('2024-03-04 12:50:01.000', '32') `test`.b2 using `test`.`b`(key) tags('2') (time, task_id) values ('2024-03-04 12:50:01.000', '43') `test`.b3 using `test`.`b`(key) tags('3') (time, task_id) values ('2024-03-04 12:50:01.000', '123456')") tdSql.execute(f"insert into `test`.b1 using `test`.`b`(key) tags('1') (time, task_id) values ('2024-03-04 12:50:01.000', '32') `test`.b2 using `test`.`b`(key) tags('2') (time, task_id) values ('2024-03-04 12:50:01.000', '43') `test`.b3 using `test`.`b`(key) tags('3') (time, task_id) values ('2024-03-04 12:50:01.000', '123456')")
tdSql.execute(f'create topic tt as select tbname,task_id,key from b') tdSql.execute(f'create topic tt as select tbname,task_id,key from b')
consumer_dict = { consumer_dict = {
"group.id": "g1", "group.id": "g1",
"td.connect.user": "root", "td.connect.user": "root",
@ -375,7 +376,7 @@ class TDTestCase:
consumer = Consumer(consumer_dict) consumer = Consumer(consumer_dict)
try: try:
consumer.subscribe(["tt"]) consumer.subscribe(["tt"])
except TmqError: except TmqError:
tdLog.exit(f"subscribe error") tdLog.exit(f"subscribe error")
@ -395,9 +396,36 @@ class TDTestCase:
finally: finally:
consumer.close() consumer.close()
def consume_ts_4544(self):
tdSql.execute(f'create database if not exists d1')
tdSql.execute(f'use d1')
tdSql.execute(f'create table stt(ts timestamp, i int) tags(t int)')
tdSql.execute(f'insert into tt1 using stt tags(1) values(now, 1) (now+1s, 2)')
tdSql.execute(f'insert into tt2 using stt tags(2) values(now, 1) (now+1s, 2)')
tdSql.execute(f'insert into tt3 using stt tags(3) values(now, 1) (now+1s, 2)')
tdSql.execute(f'insert into tt1 using stt tags(1) values(now+5s, 11) (now+10s, 12)')
tdSql.execute(f'create topic topic_in as select * from stt where tbname in ("tt2")')
consumer_dict = {
"group.id": "g1",
"td.connect.user": "root",
"td.connect.pass": "taosdata",
"auto.offset.reset": "earliest",
}
consumer = Consumer(consumer_dict)
try:
consumer.subscribe(["topic_in"])
except TmqError:
tdLog.exit(f"subscribe error")
consumer.close()
def run(self): def run(self):
self.consumeTest() self.consumeTest()
self.consume_ts_4544()
self.consume_TS_4540_Test() self.consume_TS_4540_Test()
tdSql.prepare() tdSql.prepare()