From bc98324c1df74e858ce0f742bbae9e01c939283d Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Fri, 3 Feb 2023 16:38:09 +0800 Subject: [PATCH 01/34] fix: correct ctg job error code --- source/libs/catalog/src/ctgAsync.c | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c index 325d6e0e46..f84f8e2917 100644 --- a/source/libs/catalog/src/ctgAsync.c +++ b/source/libs/catalog/src/ctgAsync.c @@ -906,9 +906,14 @@ int32_t ctgCallUserCb(void* param) { } void ctgUpdateJobErrCode(SCtgJob* pJob, int32_t errCode) { - if (!NEED_CLIENT_REFRESH_VG_ERROR(errCode) || errCode == TSDB_CODE_SUCCESS) return; + if (errCode == TSDB_CODE_SUCCESS) return; - atomic_store_32(&pJob->jobResCode, errCode); + if (NEED_CLIENT_HANDLE_ERROR(errCode)) { + atomic_store_32(&pJob->jobResCode, errCode); + } else if (0 != atomic_val_compare_exchange_32(&pJob->jobResCode, 0, errCode)) { + return; + } + qDebug("QID:0x%" PRIx64 " ctg job errCode updated to %s", pJob->queryId, tstrerror(errCode)); return; } From 6052018e77656b503ed2a62e5eb605264299f654 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 20 Feb 2023 16:42:25 +0800 Subject: [PATCH 02/34] fix:dispatch_semaphore_wait use error --- source/os/src/osSemaphore.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/source/os/src/osSemaphore.c b/source/os/src/osSemaphore.c index 2f947d3252..1f2df09ce1 100644 --- a/source/os/src/osSemaphore.c +++ b/source/os/src/osSemaphore.c @@ -132,7 +132,8 @@ int tsem_wait(tsem_t *psem) { int tsem_timewait(tsem_t *psem, int64_t milis) { if (psem == NULL || *psem == NULL) return -1; - dispatch_semaphore_wait(*psem, milis * 1000 * 1000); + dispatch_time_t time = dispatch_time(DISPATCH_TIME_NOW, (int64_t)(milis * USEC_PER_SEC)); + dispatch_semaphore_wait(*psem, time); return 0; } From 0c7a4bfad950ff93ccb9773de220a08ba515b9d1 Mon Sep 17 00:00:00 2001 From: Benguang Zhao Date: Mon, 20 Feb 2023 18:48:07 +0800 Subject: [PATCH 03/34] enh: refactor syncBeginSnapshot and walBeginSnapshot for logRetention --- include/libs/wal/wal.h | 3 +- source/dnode/mnode/sdb/src/sdbFile.c | 2 +- source/libs/sync/src/syncMain.c | 96 +++++++--------------------- source/libs/wal/src/walWrite.c | 28 ++++---- source/libs/wal/test/walMetaTest.cpp | 6 +- 5 files changed, 46 insertions(+), 89 deletions(-) diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 1eed342f8c..169013d6c1 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -66,6 +66,7 @@ typedef struct { int64_t commitVer; int64_t appliedVer; int64_t lastVer; + int64_t logRetention; } SWalVer; #pragma pack(push, 1) @@ -180,7 +181,7 @@ void walFsync(SWal *, bool force); int32_t walCommit(SWal *, int64_t ver); int32_t walRollback(SWal *, int64_t ver); // notify that previous logs can be pruned safely -int32_t walBeginSnapshot(SWal *, int64_t ver); +int32_t walBeginSnapshot(SWal *, int64_t ver, int64_t logRetention); int32_t walEndSnapshot(SWal *); int32_t walRestoreFromSnapshot(SWal *, int64_t ver); // for tq diff --git a/source/dnode/mnode/sdb/src/sdbFile.c b/source/dnode/mnode/sdb/src/sdbFile.c index c2d7a9757a..2d4b7a1e56 100644 --- a/source/dnode/mnode/sdb/src/sdbFile.c +++ b/source/dnode/mnode/sdb/src/sdbFile.c @@ -472,7 +472,7 @@ int32_t sdbWriteFile(SSdb *pSdb, int32_t delta) { taosThreadMutexLock(&pSdb->filelock); if (pSdb->pWal != NULL) { - // code = walBeginSnapshot(pSdb->pWal, pSdb->applyIndex); + // code = walBeginSnapshot(pSdb->pWal, pSdb->applyIndex, 0); if (pSdb->sync == 0) { code = 0; } else { diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 3f0432d998..1deb9fa066 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -270,88 +270,40 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) { return -1; } + SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore); + SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore); + bool isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore); + + if (isEmpty || !(lastApplyIndex >= beginIndex && lastApplyIndex <= endIndex)) { + sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 ", empty:%d, do not delete wal", lastApplyIndex, isEmpty); + syncNodeRelease(pSyncNode); + return 0; + } + int32_t code = 0; + int64_t logRetention = 0; if (syncNodeIsMnode(pSyncNode)) { // mnode - int64_t logRetention = SYNC_MNODE_LOG_RETENTION; - - SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore); - SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore); - int64_t logNum = endIndex - beginIndex; - bool isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore); - - if (isEmpty || (!isEmpty && logNum < logRetention)) { - sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 ", log-num:%" PRId64 ", empty:%d, do not delete wal", - lastApplyIndex, logNum, isEmpty); - syncNodeRelease(pSyncNode); - return 0; - } - - goto _DEL_WAL; - + logRetention = SYNC_MNODE_LOG_RETENTION; } else { - SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore); - SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore); - bool isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore); - - if (isEmpty || !(lastApplyIndex >= beginIndex && lastApplyIndex <= endIndex)) { - sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 ", empty:%d, do not delete wal", lastApplyIndex, isEmpty); - syncNodeRelease(pSyncNode); - return 0; - } - // vnode if (pSyncNode->replicaNum > 1) { // multi replicas - - lastApplyIndex = TMAX(lastApplyIndex - SYNC_VNODE_LOG_RETENTION, beginIndex - 1); - - if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { - pSyncNode->minMatchIndex = syncMinMatchIndex(pSyncNode); - - for (int32_t i = 0; i < pSyncNode->peersNum; ++i) { - int64_t matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[i])); - if (lastApplyIndex > matchIndex) { - sNTrace(pSyncNode, - "new-snapshot-index:%" PRId64 " is greater than match-index:%" PRId64 - " of dnode:%d, do not delete wal", - lastApplyIndex, matchIndex, DID(&pSyncNode->peersId[i])); - - syncNodeRelease(pSyncNode); - return 0; - } - } - - } else if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) { - if (lastApplyIndex > pSyncNode->minMatchIndex) { - sNTrace(pSyncNode, - "new-snapshot-index:%" PRId64 " is greater than min-match-index:%" PRId64 ", do not delete wal", - lastApplyIndex, pSyncNode->minMatchIndex); - syncNodeRelease(pSyncNode); - return 0; - } - - } else if (pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE) { - sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 " candidate, do not delete wal", lastApplyIndex); - syncNodeRelease(pSyncNode); - return 0; - - } else { - sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 " unknown state, do not delete wal", lastApplyIndex); - syncNodeRelease(pSyncNode); - return 0; - } - - goto _DEL_WAL; - - } else { - // one replica - - goto _DEL_WAL; + logRetention = SYNC_VNODE_LOG_RETENTION; } } + if (pSyncNode->replicaNum > 1) { + if (pSyncNode->state != TAOS_SYNC_STATE_LEADER && pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) { + sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 " candidate or unknown state, do not delete wal", + lastApplyIndex); + syncNodeRelease(pSyncNode); + return 0; + } + logRetention = TMAX(logRetention, lastApplyIndex - pSyncNode->minMatchIndex); + } + _DEL_WAL: do { @@ -366,7 +318,7 @@ _DEL_WAL: atomic_store_64(&pSyncNode->snapshottingIndex, lastApplyIndex); pSyncNode->snapshottingTime = taosGetTimestampMs(); - code = walBeginSnapshot(pData->pWal, lastApplyIndex); + code = walBeginSnapshot(pData->pWal, lastApplyIndex, logRetention); if (code == 0) { sNTrace(pSyncNode, "wal snapshot begin, index:%" PRId64 ", last apply index:%" PRId64, pSyncNode->snapshottingIndex, lastApplyIndex); diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 96c77d0971..b38961709e 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -247,21 +247,23 @@ static FORCE_INLINE int32_t walCheckAndRoll(SWal *pWal) { return 0; } -int32_t walBeginSnapshot(SWal *pWal, int64_t ver) { +int32_t walBeginSnapshot(SWal *pWal, int64_t ver, int64_t logRetention) { taosThreadMutexLock(&pWal->mutex); - + ASSERT(logRetention >= 0); pWal->vers.verInSnapshotting = ver; - wDebug("vgId:%d, wal begin snapshot for version %" PRId64 ", first ver %" PRId64 ", last ver %" PRId64, - pWal->cfg.vgId, ver, pWal->vers.firstVer, pWal->vers.lastVer); + pWal->vers.logRetention = logRetention; + + wDebug("vgId:%d, wal begin snapshot for version %" PRId64 ", log retention %" PRId64 " first ver %" PRId64 + ", last ver %" PRId64, + pWal->cfg.vgId, ver, pWal->vers.logRetention, pWal->vers.firstVer, pWal->vers.lastVer); // check file rolling - if (pWal->cfg.retentionPeriod == 0) { - if (walGetLastFileSize(pWal) != 0) { - if (walRollImpl(pWal) < 0) { - wError("vgId:%d, failed to roll wal files since %s", pWal->cfg.vgId, terrstr()); - goto _err; - } + if (walGetLastFileSize(pWal) != 0) { + if (walRollImpl(pWal) < 0) { + wError("vgId:%d, failed to roll wal files since %s", pWal->cfg.vgId, terrstr()); + goto _err; } } + taosThreadMutexUnlock(&pWal->mutex); return 0; @@ -275,8 +277,9 @@ int32_t walEndSnapshot(SWal *pWal) { taosThreadMutexLock(&pWal->mutex); int64_t ver = pWal->vers.verInSnapshotting; - wDebug("vgId:%d, wal end snapshot for version %" PRId64 ", first ver %" PRId64 ", last ver %" PRId64, pWal->cfg.vgId, - ver, pWal->vers.firstVer, pWal->vers.lastVer); + wDebug("vgId:%d, wal end snapshot for version %" PRId64 ", log retention %" PRId64 " first ver %" PRId64 + ", last ver %" PRId64, + pWal->cfg.vgId, ver, pWal->vers.logRetention, pWal->vers.firstVer, pWal->vers.lastVer); if (ver == -1) { code = -1; @@ -286,6 +289,7 @@ int32_t walEndSnapshot(SWal *pWal) { pWal->vers.snapshotVer = ver; int ts = taosGetTimestampSec(); + ver = TMAX(ver - pWal->vers.logRetention, pWal->vers.firstVer - 1); void *pIter = NULL; while (1) { pIter = taosHashIterate(pWal->pRefHash, pIter); diff --git a/source/libs/wal/test/walMetaTest.cpp b/source/libs/wal/test/walMetaTest.cpp index 891e7dcdae..0784db917a 100644 --- a/source/libs/wal/test/walMetaTest.cpp +++ b/source/libs/wal/test/walMetaTest.cpp @@ -264,7 +264,7 @@ TEST_F(WalCleanEnv, rollbackMultiFile) { ASSERT_EQ(code, 0); ASSERT_EQ(pWal->vers.lastVer, i); if (i == 5) { - walBeginSnapshot(pWal, i); + walBeginSnapshot(pWal, i, 0); walEndSnapshot(pWal); } } @@ -301,7 +301,7 @@ TEST_F(WalCleanDeleteEnv, roll) { ASSERT_EQ(pWal->vers.commitVer, i); } - walBeginSnapshot(pWal, i - 1); + walBeginSnapshot(pWal, i - 1, 0); ASSERT_EQ(pWal->vers.verInSnapshotting, i - 1); walEndSnapshot(pWal); ASSERT_EQ(pWal->vers.snapshotVer, i - 1); @@ -317,7 +317,7 @@ TEST_F(WalCleanDeleteEnv, roll) { ASSERT_EQ(pWal->vers.commitVer, i); } - code = walBeginSnapshot(pWal, i - 1); + code = walBeginSnapshot(pWal, i - 1, 0); ASSERT_EQ(code, 0); code = walEndSnapshot(pWal); ASSERT_EQ(code, 0); From ad56390a08439de012730cfb15451d8b9041d14c Mon Sep 17 00:00:00 2001 From: cyang Date: Tue, 21 Feb 2023 00:54:34 +0800 Subject: [PATCH 04/34] fix:compare func for topic & doublue free in taosArrayRemoveDuplicateP for topic --- source/dnode/mnode/impl/src/mndConsumer.c | 2 +- source/util/src/tarray.c | 5 +++-- source/util/src/tcompare.c | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 1aa2fa997b..4bad0e8563 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -626,7 +626,7 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) { } else { char *oldTopic = taosArrayGetP(pConsumerOld->currentTopics, i); char *newTopic = taosArrayGetP(newSub, j); - int comp = compareLenPrefixedStr(oldTopic, newTopic); + int comp = strcmp(oldTopic, newTopic); if (comp == 0) { i++; j++; diff --git a/source/util/src/tarray.c b/source/util/src/tarray.c index 64701574bb..8da9746b02 100644 --- a/source/util/src/tarray.c +++ b/source/util/src/tarray.c @@ -133,13 +133,14 @@ void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp) // do nothing } else { if (pos + 1 != i) { - void* p = taosArrayGet(pArray, pos + 1); + void* p = taosArrayGetp(pArray, pos + 1); if (fp != NULL) { fp(p); } taosArraySet(pArray, pos + 1, p2); - pos += 1; + memset(TARRAY_GET_ELEM(pArray, i), 0, pArray->elemSize); + pos += 1; } else { pos += 1; } diff --git a/source/util/src/tcompare.c b/source/util/src/tcompare.c index 7a52991e81..46a9051436 100644 --- a/source/util/src/tcompare.c +++ b/source/util/src/tcompare.c @@ -1232,7 +1232,7 @@ int32_t taosArrayCompareString(const void *a, const void *b) { const char *x = *(const char **)a; const char *y = *(const char **)b; - return compareLenPrefixedStr(x, y); + return strcmp(x, y); } int32_t comparestrPatternMatch(const void *pLeft, const void *pRight) { From 968c538be5beb0428307e8729a329b21a34e2d23 Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Tue, 21 Feb 2023 08:54:05 +0800 Subject: [PATCH 05/34] fix: add merge join operator result limitation --- source/common/src/tdatablock.c | 3 + source/libs/executor/src/joinoperator.c | 134 +++++++++++++++++++----- tests/parallel_test/cases.task | 1 + tests/script/tsim/query/join.sim | 72 +++++++++++++ 4 files changed, 181 insertions(+), 29 deletions(-) create mode 100644 tests/script/tsim/query/join.sim diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 86dcd1eceb..b8bd6c09bc 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -94,6 +94,9 @@ int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, con while (newSize < pAttr->length + dataLen) { newSize = newSize * 1.5; + if (newSize > UINT32_MAX) { + return TSDB_CODE_OUT_OF_MEMORY; + } } char* buf = taosMemoryRealloc(pColumnInfoData->pData, newSize); diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 88ed9eccb3..7a2277ed9c 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -24,6 +24,17 @@ #include "tmsg.h" #include "ttypes.h" +typedef struct SJoinRowCtx { + bool rowRemains; + int64_t ts; + SArray* leftRowLocations; + SArray* rightRowLocations; + SArray* leftCreatedBlocks; + SArray* rightCreatedBlocks; + int32_t leftRowIdx; + int32_t rightRowIdx; +} SJoinRowCtx; + typedef struct SJoinOperatorInfo { SSDataBlock* pRes; int32_t joinType; @@ -37,6 +48,8 @@ typedef struct SJoinOperatorInfo { int32_t rightPos; SColumnInfo rightCol; SNode* pCondAfterMerge; + + SJoinRowCtx rowCtx; } SJoinOperatorInfo; static void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode); @@ -287,49 +300,107 @@ static int32_t mergeJoinGetDownStreamRowsEqualTimeStamp(SOperatorInfo* pOperator static int32_t mergeJoinJoinDownstreamTsRanges(SOperatorInfo* pOperator, int64_t timestamp, SSDataBlock* pRes, int32_t* nRows) { - SJoinOperatorInfo* pJoinInfo = pOperator->info; - SArray* leftRowLocations = taosArrayInit(8, sizeof(SRowLocation)); - SArray* leftCreatedBlocks = taosArrayInit(8, POINTER_BYTES); - - SArray* rightRowLocations = taosArrayInit(8, sizeof(SRowLocation)); - SArray* rightCreatedBlocks = taosArrayInit(8, POINTER_BYTES); int32_t code = TSDB_CODE_SUCCESS; - mergeJoinGetDownStreamRowsEqualTimeStamp(pOperator, 0, pJoinInfo->leftCol.slotId, pJoinInfo->pLeft, - pJoinInfo->leftPos, timestamp, leftRowLocations, leftCreatedBlocks); - mergeJoinGetDownStreamRowsEqualTimeStamp(pOperator, 1, pJoinInfo->rightCol.slotId, pJoinInfo->pRight, - pJoinInfo->rightPos, timestamp, rightRowLocations, rightCreatedBlocks); + SJoinOperatorInfo* pJoinInfo = pOperator->info; + SArray* leftRowLocations = NULL; + SArray* leftCreatedBlocks = NULL; + SArray* rightRowLocations = NULL; + SArray* rightCreatedBlocks = NULL; + int32_t leftRowIdx = 0; + int32_t rightRowIdx = 0; + int32_t i, j; + + if (pJoinInfo->rowCtx.rowRemains) { + leftRowLocations = pJoinInfo->rowCtx.leftRowLocations; + leftCreatedBlocks = pJoinInfo->rowCtx.leftCreatedBlocks; + rightRowLocations = pJoinInfo->rowCtx.rightRowLocations; + rightCreatedBlocks = pJoinInfo->rowCtx.rightCreatedBlocks; + leftRowIdx = pJoinInfo->rowCtx.leftRowIdx; + rightRowIdx = pJoinInfo->rowCtx.rightRowIdx; + } else { + leftRowLocations = taosArrayInit(8, sizeof(SRowLocation)); + leftCreatedBlocks = taosArrayInit(8, POINTER_BYTES); + rightRowLocations = taosArrayInit(8, sizeof(SRowLocation)); + rightCreatedBlocks = taosArrayInit(8, POINTER_BYTES); + + mergeJoinGetDownStreamRowsEqualTimeStamp(pOperator, 0, pJoinInfo->leftCol.slotId, pJoinInfo->pLeft, + pJoinInfo->leftPos, timestamp, leftRowLocations, leftCreatedBlocks); + mergeJoinGetDownStreamRowsEqualTimeStamp(pOperator, 1, pJoinInfo->rightCol.slotId, pJoinInfo->pRight, + pJoinInfo->rightPos, timestamp, rightRowLocations, rightCreatedBlocks); + } + size_t leftNumJoin = taosArrayGetSize(leftRowLocations); size_t rightNumJoin = taosArrayGetSize(rightRowLocations); - code = blockDataEnsureCapacity(pRes, *nRows + leftNumJoin * rightNumJoin); + uint32_t maxRowNum = *nRows + (leftNumJoin - leftRowIdx - 1) * rightNumJoin + rightNumJoin - rightRowIdx; + uint32_t limitRowNum = maxRowNum; + if (maxRowNum > pOperator->resultInfo.threshold) { + limitRowNum = pOperator->resultInfo.threshold; + if (!pJoinInfo->rowCtx.rowRemains) { + pJoinInfo->rowCtx.rowRemains = true; + pJoinInfo->rowCtx.ts = timestamp; + pJoinInfo->rowCtx.leftRowLocations = leftRowLocations; + pJoinInfo->rowCtx.rightRowLocations = rightRowLocations; + pJoinInfo->rowCtx.leftCreatedBlocks = leftCreatedBlocks; + pJoinInfo->rowCtx.rightCreatedBlocks = rightCreatedBlocks; + } + } + + code = blockDataEnsureCapacity(pRes, limitRowNum); if (code != TSDB_CODE_SUCCESS) { qError("%s can not ensure block capacity for join. left: %zu, right: %zu", GET_TASKID(pOperator->pTaskInfo), leftNumJoin, rightNumJoin); } + + if (code == TSDB_CODE_SUCCESS) { - for (int32_t i = 0; i < leftNumJoin; ++i) { - for (int32_t j = 0; j < rightNumJoin; ++j) { + bool done = false; + for (i = leftRowIdx; i < leftNumJoin; ++i, rightRowIdx = 0) { + for (j = rightRowIdx; j < rightNumJoin; ++j) { + if (*nRows >= limitRowNum) { + done = true; + break; + } + SRowLocation* leftRow = taosArrayGet(leftRowLocations, i); SRowLocation* rightRow = taosArrayGet(rightRowLocations, j); mergeJoinJoinLeftRight(pOperator, pRes, *nRows, leftRow->pDataBlock, leftRow->pos, rightRow->pDataBlock, rightRow->pos); ++*nRows; } + if (done) { + break; + } + } + + if (maxRowNum > pOperator->resultInfo.threshold) { + pJoinInfo->rowCtx.leftRowIdx = i; + pJoinInfo->rowCtx.rightRowIdx = j; } } - for (int i = 0; i < taosArrayGetSize(rightCreatedBlocks); ++i) { - SSDataBlock* pBlock = taosArrayGetP(rightCreatedBlocks, i); - blockDataDestroy(pBlock); + if (maxRowNum <= pOperator->resultInfo.threshold) { + for (int i = 0; i < taosArrayGetSize(rightCreatedBlocks); ++i) { + SSDataBlock* pBlock = taosArrayGetP(rightCreatedBlocks, i); + blockDataDestroy(pBlock); + } + taosArrayDestroy(rightCreatedBlocks); + taosArrayDestroy(rightRowLocations); + for (int i = 0; i < taosArrayGetSize(leftCreatedBlocks); ++i) { + SSDataBlock* pBlock = taosArrayGetP(leftCreatedBlocks, i); + blockDataDestroy(pBlock); + } + taosArrayDestroy(leftCreatedBlocks); + taosArrayDestroy(leftRowLocations); + + if (pJoinInfo->rowCtx.rowRemains) { + pJoinInfo->rowCtx.rowRemains = false; + pJoinInfo->rowCtx.leftRowLocations = NULL; + pJoinInfo->rowCtx.rightRowLocations = NULL; + pJoinInfo->rowCtx.leftCreatedBlocks = NULL; + pJoinInfo->rowCtx.rightCreatedBlocks = NULL; + } } - taosArrayDestroy(rightCreatedBlocks); - taosArrayDestroy(rightRowLocations); - for (int i = 0; i < taosArrayGetSize(leftCreatedBlocks); ++i) { - SSDataBlock* pBlock = taosArrayGetP(leftCreatedBlocks, i); - blockDataDestroy(pBlock); - } - taosArrayDestroy(leftCreatedBlocks); - taosArrayDestroy(leftRowLocations); return TSDB_CODE_SUCCESS; } @@ -379,9 +450,14 @@ static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) while (1) { int64_t leftTs = 0; int64_t rightTs = 0; - bool hasNextTs = mergeJoinGetNextTimestamp(pOperator, &leftTs, &rightTs); - if (!hasNextTs) { - break; + if (pJoinInfo->rowCtx.rowRemains) { + leftTs = pJoinInfo->rowCtx.ts; + rightTs = pJoinInfo->rowCtx.ts; + } else { + bool hasNextTs = mergeJoinGetNextTimestamp(pOperator, &leftTs, &rightTs); + if (!hasNextTs) { + break; + } } if (leftTs == rightTs) { @@ -389,12 +465,12 @@ static void doMergeJoinImpl(struct SOperatorInfo* pOperator, SSDataBlock* pRes) } else if ((asc && leftTs < rightTs) || (!asc && leftTs > rightTs)) { pJoinInfo->leftPos += 1; - if (pJoinInfo->leftPos >= pJoinInfo->pLeft->info.rows) { + if (pJoinInfo->leftPos >= pJoinInfo->pLeft->info.rows && pRes->info.rows < pOperator->resultInfo.threshold) { continue; } } else if ((asc && leftTs > rightTs) || (!asc && leftTs < rightTs)) { pJoinInfo->rightPos += 1; - if (pJoinInfo->rightPos >= pJoinInfo->pRight->info.rows) { + if (pJoinInfo->rightPos >= pJoinInfo->pRight->info.rows && pRes->info.rows < pOperator->resultInfo.threshold) { continue; } } diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 16751423b1..af3cd6206b 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -179,6 +179,7 @@ ,,y,script,./test.sh -f tsim/query/sys_tbname.sim ,,y,script,./test.sh -f tsim/query/groupby.sim ,,y,script,./test.sh -f tsim/query/forceFill.sim +,,y,script,./test.sh -f tsim/query/join.sim ,,y,script,./test.sh -f tsim/qnode/basic1.sim ,,y,script,./test.sh -f tsim/snode/basic1.sim ,,y,script,./test.sh -f tsim/mnode/basic1.sim diff --git a/tests/script/tsim/query/join.sim b/tests/script/tsim/query/join.sim new file mode 100644 index 0000000000..adb0338ef7 --- /dev/null +++ b/tests/script/tsim/query/join.sim @@ -0,0 +1,72 @@ +system sh/stop_dnodes.sh +system sh/deploy.sh -n dnode1 -i 1 +system sh/exec.sh -n dnode1 -s start +sql connect + +$dbPrefix = db +$tbPrefix1 = tba +$tbPrefix2 = tbb +$mtPrefix = stb +$tbNum = 10000 +$rowNum = 2 + +print =============== step1 +$i = 0 +$db = $dbPrefix . $i +$mt1 = $mtPrefix . $i +$i = 1 +$mt2 = $mtPrefix . $i + +sql drop database $db -x step1 +step1: +sql create database $db +sql use $db +sql create table $mt1 (ts timestamp, f1 int) TAGS(tag1 int, tag2 binary(500)) +sql create table $mt2 (ts timestamp, f1 int) TAGS(tag1 int, tag2 binary(500)) + +print ====== start create child tables and insert data +$i = 0 +while $i < $tbNum + $tb = $tbPrefix1 . $i + sql create table $tb using $mt1 tags( $i , 'aaaaaaaaaaaaaaaaaaaaaaaaaaa') + + $x = 0 + while $x < $rowNum + $cc = $x * 60000 + $ms = 1601481600000 + $cc + + sql insert into $tb values ($ms , $x ) + $x = $x + 1 + endw + + $i = $i + 1 +endw + +print =============== step2 +$i = 0 +while $i < $tbNum + $tb = $tbPrefix2 . $i + sql create table $tb using $mt2 tags( $i , 'aaaaaaaaaaaaaaaaaaaaaaaaaaa') + + $x = 0 + while $x < $rowNum + $cc = $x * 60000 + $ms = 1601481600000 + $cc + + sql insert into $tb values ($ms , $x ) + $x = $x + 1 + endw + + $i = $i + 1 +endw + +sql select * from tba0 t1, tbb0 t2 where t1.ts=t2.ts; +if $rows != 2 then + return -1 +endi +sql select * from stb0 t1, stb1 t2 where t1.ts=t2.ts and t1.tag2=t2.tag2; +if $rows != 200000000 then + return -1 +endi + +system sh/exec.sh -n dnode1 -s stop -x SIGINT From f31dd176314b210a6a861bb5042d8688ce76a815 Mon Sep 17 00:00:00 2001 From: cyang Date: Tue, 21 Feb 2023 11:04:18 +0800 Subject: [PATCH 06/34] fix:compare func for topic & doublue free in taosArrayRemoveDuplicateP for topic --- source/util/src/tarray.c | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/source/util/src/tarray.c b/source/util/src/tarray.c index 8da9746b02..6c951da9e1 100644 --- a/source/util/src/tarray.c +++ b/source/util/src/tarray.c @@ -133,7 +133,7 @@ void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp) // do nothing } else { if (pos + 1 != i) { - void* p = taosArrayGetp(pArray, pos + 1); + void* p = taosArrayGet(pArray, pos + 1); if (fp != NULL) { fp(p); } @@ -172,13 +172,14 @@ void taosArrayRemoveDuplicateP(SArray* pArray, __compar_fn_t comparFn, void (*fp // do nothing } else { if (pos + 1 != i) { - void* p = taosArrayGet(pArray, pos + 1); + void* p = taosArrayGetP(pArray, pos + 1); if (fp != NULL) { fp(p); } taosArraySet(pArray, pos + 1, p2); - pos += 1; + memset(TARRAY_GET_ELEM(pArray, i), 0, pArray->elemSize); + pos += 1; } else { pos += 1; } From 6e0a349423bd191941d022b56ffb71982613a09f Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Tue, 21 Feb 2023 11:40:33 +0800 Subject: [PATCH 07/34] fix: remove case from ci cause of long exectuion time --- tests/parallel_test/cases.task | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index af3cd6206b..437abf4917 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -179,7 +179,7 @@ ,,y,script,./test.sh -f tsim/query/sys_tbname.sim ,,y,script,./test.sh -f tsim/query/groupby.sim ,,y,script,./test.sh -f tsim/query/forceFill.sim -,,y,script,./test.sh -f tsim/query/join.sim +,,n,script,./test.sh -f tsim/query/join.sim ,,y,script,./test.sh -f tsim/qnode/basic1.sim ,,y,script,./test.sh -f tsim/snode/basic1.sim ,,y,script,./test.sh -f tsim/mnode/basic1.sim From 3df3fea0bdf5666c1e793f3b54e4182e6a9d279a Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Tue, 21 Feb 2023 13:29:04 +0800 Subject: [PATCH 08/34] fix: tag filter crash issue --- source/libs/executor/src/executil.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 040e67713d..a7019d047f 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -873,13 +873,14 @@ static SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTa } else if (pColInfo->info.type == TSDB_DATA_TYPE_JSON) { colDataAppend(pColInfo, i, p, false); } else if (IS_VAR_DATA_TYPE(pColInfo->info.type)) { - char* tmp = alloca(tagVal.nData + VARSTR_HEADER_SIZE + 1); + char* tmp = taosMemoryMalloc(tagVal.nData + VARSTR_HEADER_SIZE + 1); varDataSetLen(tmp, tagVal.nData); memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData); colDataAppend(pColInfo, i, tmp, false); #if TAG_FILTER_DEBUG qDebug("tagfilter varch:%s", tmp + 2); #endif + taosMemoryFree(tmp); } else { colDataAppend(pColInfo, i, (const char*)&tagVal.i64, false); #if TAG_FILTER_DEBUG From e5023259ffb53682fa3df26b19170258d58dd243 Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Tue, 21 Feb 2023 14:36:25 +0800 Subject: [PATCH 09/34] fix: add tag filter case --- tests/parallel_test/cases.task | 1 + tests/system-test/2-query/tagFilter.py | 67 ++++++++++++++++++++++++++ 2 files changed, 68 insertions(+) create mode 100644 tests/system-test/2-query/tagFilter.py diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 16751423b1..50016dd963 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -646,6 +646,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/elapsed.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/csum.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/function_diff.py +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tagFilter.py ,,n,system-test,python3 ./test.py -f 2-query/queryQnode.py ,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode1mnode.py ,,y,system-test,./pytest.sh python3 ./test.py -f 6-cluster/5dnode2mnode.py -N 5 diff --git a/tests/system-test/2-query/tagFilter.py b/tests/system-test/2-query/tagFilter.py new file mode 100644 index 0000000000..b03776c31b --- /dev/null +++ b/tests/system-test/2-query/tagFilter.py @@ -0,0 +1,67 @@ +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * + + +INT_COL = "c1" +BINT_COL = "c2" +SINT_COL = "c3" +TINT_COL = "c4" +FLOAT_COL = "c5" +DOUBLE_COL = "c6" +BOOL_COL = "c7" + +BINARY_COL = "c8" +NCHAR_COL = "c9" +TS_COL = "c10" + +NUM_COL = [INT_COL, BINT_COL, SINT_COL, TINT_COL, FLOAT_COL, DOUBLE_COL, ] +UN_NUM_COL = [BOOL_COL, BINARY_COL, NCHAR_COL, ] +TS_TYPE_COL = [TS_COL] + +DBNAME = "db" + +class TDTestCase: + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor()) + tdSql.execute(f'drop database if exists db') + tdSql.execute(f'create database if not exists db vgroups 1') + + def __create_tb(self, dbname="db"): + create_stb_sql = f'''create table {dbname}.stb1( + ts timestamp, f1 int + ) tags (tag1 binary(16300)) + ''' + tdSql.execute(create_stb_sql) + + tag_value = 'a' + for i in range(1200): + tag_value = tag_value + 'a' + + for i in range(8000): + tdSql.execute(f"create table {dbname}.ct{i+1} using {dbname}.stb1 tags ( '{tag_value}' )") + + def __query_data(self, rows, dbname="db"): + tdSql.execute( + f'''select count(*) from {dbname}.stb1 where tag1 like '%a' + ''' + ) + tdSql.checkRows(0) + + def run(self): + tdLog.printNoPrefix("==========step1:create table") + self.__create_tb() + + tdLog.printNoPrefix("==========step2:query data") + self.__query_data(10) + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) From 4e0d69c6b00df7175501742089e7d886e5483eee Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Tue, 21 Feb 2023 16:11:14 +0800 Subject: [PATCH 10/34] fix: memory leak in tdb --- source/libs/tdb/src/db/tdbBtree.c | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c index 4f0682a617..a35e01c7aa 100644 --- a/source/libs/tdb/src/db/tdbBtree.c +++ b/source/libs/tdb/src/db/tdbBtree.c @@ -2188,6 +2188,9 @@ int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) { } else { lidx = lidx + 1; } + if (TDB_CELLDECODER_FREE_KEY(&pBtc->coder)) { + tdbFree((void*)pTKey); + } // compare last cell if (lidx <= ridx) { @@ -2199,6 +2202,9 @@ int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) { } else { ridx = ridx - 1; } + if (TDB_CELLDECODER_FREE_KEY(&pBtc->coder)) { + tdbFree((void*)pTKey); + } } // binary search @@ -2209,6 +2215,9 @@ int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) { pBtc->idx = (lidx + ridx) >> 1; tdbBtcGet(pBtc, &pTKey, &tkLen, NULL, NULL); c = pBt->kcmpr(pKey, kLen, pTKey, tkLen); + if (TDB_CELLDECODER_FREE_KEY(&pBtc->coder)) { + tdbFree((void*)pTKey); + } if (c < 0) { // pKey < cd.pKey ridx = pBtc->idx - 1; From 6bb9ee9b4eed35a3c5d411b60a26c031521aba29 Mon Sep 17 00:00:00 2001 From: Benguang Zhao Date: Mon, 20 Feb 2023 13:51:14 +0800 Subject: [PATCH 11/34] enh: change the default of vndCommitMaxInterval as 10 minutes --- source/common/src/tglobal.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index e3f08e912a..c8f1efa1ab 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -61,7 +61,7 @@ int32_t tsHeartbeatInterval = 1000; int32_t tsHeartbeatTimeout = 20 * 1000; // vnode -int64_t tsVndCommitMaxIntervalMs = 60 * 1000; +int64_t tsVndCommitMaxIntervalMs = 600 * 1000; // monitor bool tsEnableMonitor = true; From a78556a3e66058bb47aa7b4455a827b8723b8ac7 Mon Sep 17 00:00:00 2001 From: Benguang Zhao Date: Tue, 21 Feb 2023 16:48:15 +0800 Subject: [PATCH 12/34] fix: synchronize access to pVnode->inUse --- source/dnode/vnode/src/vnd/vnodeBufPool.c | 7 ++++++- source/dnode/vnode/src/vnd/vnodeCommit.c | 24 +++++++++++------------ 2 files changed, 17 insertions(+), 14 deletions(-) diff --git a/source/dnode/vnode/src/vnd/vnodeBufPool.c b/source/dnode/vnode/src/vnd/vnodeBufPool.c index 83a414dae0..88abc1b3f0 100644 --- a/source/dnode/vnode/src/vnd/vnodeBufPool.c +++ b/source/dnode/vnode/src/vnd/vnodeBufPool.c @@ -94,6 +94,7 @@ int vnodeOpenBufPool(SVnode *pVnode) { int vnodeCloseBufPool(SVnode *pVnode) { SVBufPool *pPool; + taosThreadMutexLock(&pVnode->mutex); for (pPool = pVnode->pPool; pPool; pPool = pVnode->pPool) { pVnode->pPool = pPool->next; vnodeBufPoolDestroy(pPool); @@ -103,8 +104,9 @@ int vnodeCloseBufPool(SVnode *pVnode) { vnodeBufPoolDestroy(pVnode->inUse); pVnode->inUse = NULL; } - vDebug("vgId:%d, vnode buffer pool is closed", TD_VID(pVnode)); + taosThreadMutexUnlock(&pVnode->mutex); + vDebug("vgId:%d, vnode buffer pool is closed", TD_VID(pVnode)); return 0; } @@ -244,6 +246,9 @@ void vnodeBufPoolUnRef(SVBufPool *pPool) { pVnode->pPool = pPool; taosThreadCondSignal(&pVnode->poolNotEmpty); + if (pVnode->inUse == pPool) { + pVnode->inUse = NULL; + } taosThreadMutexUnlock(&pVnode->mutex); } } diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c index cefa9e6755..e8280ea751 100644 --- a/source/dnode/vnode/src/vnd/vnodeCommit.c +++ b/source/dnode/vnode/src/vnd/vnodeCommit.c @@ -87,22 +87,21 @@ void vnodeUpdCommitSched(SVnode *pVnode) { } int vnodeShouldCommit(SVnode *pVnode) { - if (!pVnode->inUse || !osDataSpaceAvailable()) { - return false; - } - SVCommitSched *pSched = &pVnode->commitSched; int64_t nowMs = taosGetMonoTimestampMs(); + bool diskAvail = osDataSpaceAvailable(); + bool needCommit = false; - return (((pVnode->inUse->size > pVnode->inUse->node.size) && (pSched->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs)) || - (pVnode->inUse->size > 0 && pSched->commitMs + pSched->maxWaitMs < nowMs)); -} - -int vnodeShouldCommitOld(SVnode *pVnode) { - if (pVnode->inUse) { - return osDataSpaceAvailable() && (pVnode->inUse->size > pVnode->inUse->node.size); + taosThreadMutexLock(&pVnode->mutex); + if (!pVnode->inUse || !diskAvail) { + goto _out; } - return false; + needCommit = + (((pVnode->inUse->size > pVnode->inUse->node.size) && (pSched->commitMs + SYNC_VND_COMMIT_MIN_MS < nowMs)) || + (pVnode->inUse->size > 0 && pSched->commitMs + pSched->maxWaitMs < nowMs)); +_out: + taosThreadMutexUnlock(&pVnode->mutex); + return needCommit; } int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) { @@ -259,7 +258,6 @@ static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) { metaPrepareAsyncCommit(pVnode->pMeta); vnodeBufPoolUnRef(pVnode->inUse); - pVnode->inUse = NULL; _exit: if (code) { From b28c0ad9529a3f948efda3d9c45cf2c1e2a32a20 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Tue, 21 Feb 2023 18:49:39 +0800 Subject: [PATCH 13/34] fix:remove useless logic in tmq --- examples/c/tmq.c | 2 +- include/libs/wal/wal.h | 2 +- source/dnode/vnode/src/tq/tqRead.c | 15 ++++++------ source/libs/executor/src/executor.c | 9 ------- source/libs/executor/src/scanoperator.c | 1 - source/libs/wal/src/walRead.c | 13 ++-------- source/libs/wal/src/walRef.c | 32 ++++++++++++------------- 7 files changed, 26 insertions(+), 48 deletions(-) diff --git a/examples/c/tmq.c b/examples/c/tmq.c index eb41ad039a..266acbe820 100644 --- a/examples/c/tmq.c +++ b/examples/c/tmq.c @@ -61,7 +61,7 @@ static int32_t init_env() { printf("create database\n"); pRes = taos_query(pConn, "drop topic topicname"); if (taos_errno(pRes) != 0) { - printf("error in drop tmqdb, reason:%s\n", taos_errstr(pRes)); + printf("error in drop topicname, reason:%s\n", taos_errstr(pRes)); } taos_free_result(pRes); diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h index 1eed342f8c..09d737fe93 100644 --- a/include/libs/wal/wal.h +++ b/include/libs/wal/wal.h @@ -126,7 +126,7 @@ typedef struct SWal { typedef struct { int64_t refId; int64_t refVer; - int64_t refFile; +// int64_t refFile; SWal *pWal; } SWalRef; diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index 46b31bc5b0..7cad739ffa 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -297,11 +297,8 @@ void tqCloseReader(STqReader* pReader) { int32_t tqSeekVer(STqReader* pReader, int64_t ver) { if (walReadSeekVer(pReader->pWalReader, ver) < 0) { - ASSERT(pReader->pWalReader->curInvalid); - ASSERT(pReader->pWalReader->curVersion == ver); return -1; } - ASSERT(pReader->pWalReader->curVersion == ver); return 0; } @@ -362,11 +359,13 @@ int32_t tqNextBlock(STqReader* pReader, SFetchRet* ret) { int32_t tqReaderSetDataMsg(STqReader* pReader, const SSubmitReq* pMsg, int64_t ver) { pReader->pMsg = pMsg; - if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1; - while (true) { - if (tGetSubmitMsgNext(&pReader->msgIter, &pReader->pBlock) < 0) return -1; - if (pReader->pBlock == NULL) break; - } +// if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1; +// while (true) { +// if (tGetSubmitMsgNext(&pReader->msgIter, &pReader->pBlock) < 0) return -1; +// tqDebug("submitnext vgId:%d, block:%p, dataLen:%d, len:%d, uid:%"PRId64, pReader->pWalReader->pWal->cfg.vgId, pReader->pBlock, pReader->msgIter.dataLen, +// pReader->msgIter.len, pReader->msgIter.uid); +// if (pReader->pBlock == NULL) break; +// } if (tInitSubmitMsgIter(pMsg, &pReader->msgIter) < 0) return -1; pReader->ver = ver; diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 6c354c3d61..73878cd00c 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -1049,18 +1049,9 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT STableScanInfo* pTSInfo = pInfo->pTableScanOp->info; tsdbReaderClose(pTSInfo->base.dataReader); pTSInfo->base.dataReader = NULL; -#if 0 - if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus) && - pInfo->tqReader->pWalReader->curVersion != pOffset->version) { - qError("prepare scan ver %" PRId64 " actual ver %" PRId64 ", last %" PRId64, pOffset->version, - pInfo->tqReader->pWalReader->curVersion, pTaskInfo->streamInfo.lastStatus.version); - ASSERT(0); - } -#endif if (tqSeekVer(pInfo->tqReader, pOffset->version + 1) < 0) { return -1; } - ASSERT(pInfo->tqReader->pWalReader->curVersion == pOffset->version + 1); } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) { /*pInfo->blockType = STREAM_INPUT__TABLE_SCAN;*/ int64_t uid = pOffset->uid; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index c06fc40b9b..99e630f45e 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1618,7 +1618,6 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) { tqOffsetResetToLog(&pTaskInfo->streamInfo.lastStatus, pTaskInfo->streamInfo.snapshotVer); return NULL; } - ASSERT(pInfo->tqReader->pWalReader->curVersion == pTaskInfo->streamInfo.snapshotVer + 1); } else { return NULL; } diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index 5e09af5b2e..526dba0bb5 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -96,8 +96,7 @@ int32_t walNextValidMsg(SWalReader *pReader) { if (walSkipFetchBodyNew(pReader) < 0) { return -1; } - fetchVer++; - ASSERT(fetchVer == pReader->curVersion); + fetchVer = pReader->curVersion; } } pReader->curStopped = 1; @@ -144,7 +143,7 @@ static int64_t walReadSeekFilePos(SWalReader *pReader, int64_t fileFirstVer, int } static int32_t walReadChangeFile(SWalReader *pReader, int64_t fileFirstVer) { - char fnameStr[WAL_FILE_LEN]; + char fnameStr[WAL_FILE_LEN] = {0}; taosCloseFile(&pReader->pIdxFile); taosCloseFile(&pReader->pLogFile); @@ -300,14 +299,6 @@ static int32_t walFetchBodyNew(SWalReader *pRead) { return -1; } - if (pReadHead->version != ver) { - wError("vgId:%d, wal fetch body error:%" PRId64 ", read request index:%" PRId64, pRead->pWal->cfg.vgId, - pRead->pHead->head.version, ver); - pRead->curInvalid = 1; - terrno = TSDB_CODE_WAL_FILE_CORRUPTED; - return -1; - } - if (walValidBodyCksum(pRead->pHead) != 0) { wError("vgId:%d, wal fetch body error:%" PRId64 ", since body checksum not passed", pRead->pWal->cfg.vgId, ver); pRead->curInvalid = 1; diff --git a/source/libs/wal/src/walRef.c b/source/libs/wal/src/walRef.c index 43470f4c82..768256cefa 100644 --- a/source/libs/wal/src/walRef.c +++ b/source/libs/wal/src/walRef.c @@ -26,7 +26,7 @@ SWalRef *walOpenRef(SWal *pWal) { } pRef->refId = tGenIdPI64(); pRef->refVer = -1; - pRef->refFile = -1; +// pRef->refFile = -1; pRef->pWal = pWal; taosHashPut(pWal->pRefHash, &pRef->refId, sizeof(int64_t), &pRef, sizeof(void *)); return pRef; @@ -58,11 +58,11 @@ int32_t walRefVer(SWalRef *pRef, int64_t ver) { pRef->refVer = ver; // bsearch in fileSet - SWalFileInfo tmpInfo; - tmpInfo.firstVer = ver; - SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE); - ASSERT(pRet != NULL); - pRef->refFile = pRet->firstVer; +// SWalFileInfo tmpInfo; +// tmpInfo.firstVer = ver; +// SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE); +// ASSERT(pRet != NULL); +// pRef->refFile = pRet->firstVer; taosThreadMutexUnlock(&pWal->mutex); } @@ -73,7 +73,7 @@ int32_t walRefVer(SWalRef *pRef, int64_t ver) { #if 1 void walUnrefVer(SWalRef *pRef) { pRef->refId = -1; - pRef->refFile = -1; +// pRef->refFile = -1; } #endif @@ -85,20 +85,18 @@ SWalRef *walRefFirstVer(SWal *pWal, SWalRef *pRef) { } } taosThreadMutexLock(&pWal->mutex); - int64_t ver = walGetFirstVer(pWal); - - wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, ver); - pRef->refVer = ver; // bsearch in fileSet - SWalFileInfo tmpInfo; - tmpInfo.firstVer = ver; - SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE); - ASSERT(pRet != NULL); - pRef->refFile = pRet->firstVer; +// SWalFileInfo tmpInfo; +// tmpInfo.firstVer = ver; +// SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE); +// ASSERT(pRet != NULL); +// pRef->refFile = pRet->firstVer; taosThreadMutexUnlock(&pWal->mutex); + wDebug("vgId:%d, wal ref version %" PRId64 " for first", pWal->cfg.vgId, ver); + return pRef; } @@ -119,7 +117,7 @@ SWalRef *walRefCommittedVer(SWal *pWal) { tmpInfo.firstVer = ver; SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE); ASSERT(pRet != NULL); - pRef->refFile = pRet->firstVer; +// pRef->refFile = pRet->firstVer; taosThreadMutexUnlock(&pWal->mutex); return pRef; From b0b9877265eb1a157c6d6fbb46f25977d1c81651 Mon Sep 17 00:00:00 2001 From: Benguang Zhao Date: Tue, 21 Feb 2023 19:00:47 +0800 Subject: [PATCH 14/34] fix: check if log buffer is empty in syncNodeOnLocalCmd --- source/libs/sync/inc/syncPipeline.h | 1 + source/libs/sync/src/syncMain.c | 4 ++++ source/libs/sync/src/syncPipeline.c | 11 +++++++++++ 3 files changed, 16 insertions(+) diff --git a/source/libs/sync/inc/syncPipeline.h b/source/libs/sync/inc/syncPipeline.h index 504a9f0bd7..a823cfda0b 100644 --- a/source/libs/sync/inc/syncPipeline.h +++ b/source/libs/sync/inc/syncPipeline.h @@ -99,6 +99,7 @@ int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode); // access int64_t syncLogBufferGetEndIndex(SSyncLogBuffer* pBuf); SyncTerm syncLogBufferGetLastMatchTerm(SSyncLogBuffer* pBuf); +bool syncLogBufferIsEmpty(SSyncLogBuffer* pBuf); int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEntry* pEntry); int32_t syncLogBufferAccept(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevTerm); int64_t syncLogBufferProceed(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncTerm* pMatchTerm); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 3f0432d998..07ecc63c25 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -2468,6 +2468,10 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) { syncNodeStepDown(ths, pMsg->currentTerm); } else if (pMsg->cmd == SYNC_LOCAL_CMD_FOLLOWER_CMT) { + if (syncLogBufferIsEmpty(ths->pLogBuf)) { + sError("vgId:%d, sync log buffer is empty.", ths->vgId); + return 0; + } SyncTerm matchTerm = syncLogBufferGetLastMatchTerm(ths->pLogBuf); if (pMsg->currentTerm == matchTerm) { (void)syncNodeUpdateCommitIndex(ths, pMsg->commitIndex); diff --git a/source/libs/sync/src/syncPipeline.c b/source/libs/sync/src/syncPipeline.c index c9ff2d2dcc..e2b039a2e4 100644 --- a/source/libs/sync/src/syncPipeline.c +++ b/source/libs/sync/src/syncPipeline.c @@ -253,6 +253,7 @@ int32_t syncLogBufferInit(SSyncLogBuffer* pBuf, SSyncNode* pNode) { int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode) { taosThreadMutexLock(&pBuf->mutex); + syncLogBufferValidate(pBuf); for (SyncIndex index = pBuf->startIndex; index < pBuf->endIndex; index++) { SSyncRaftEntry* pEntry = pBuf->entries[(index + pBuf->size) % pBuf->size].pItem; if (pEntry == NULL) continue; @@ -265,6 +266,7 @@ int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode) { if (ret < 0) { sError("vgId:%d, failed to re-initialize sync log buffer since %s.", pNode->vgId, terrstr()); } + syncLogBufferValidate(pBuf); taosThreadMutexUnlock(&pBuf->mutex); return ret; } @@ -283,6 +285,13 @@ SyncTerm syncLogBufferGetLastMatchTerm(SSyncLogBuffer* pBuf) { return term; } +bool syncLogBufferIsEmpty(SSyncLogBuffer* pBuf) { + taosThreadMutexLock(&pBuf->mutex); + bool empty = (pBuf->endIndex <= pBuf->startIndex); + taosThreadMutexUnlock(&pBuf->mutex); + return empty; +} + int32_t syncLogBufferAccept(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevTerm) { taosThreadMutexLock(&pBuf->mutex); syncLogBufferValidate(pBuf); @@ -1073,6 +1082,7 @@ int32_t syncLogBufferRollback(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex int32_t syncLogBufferReset(SSyncLogBuffer* pBuf, SSyncNode* pNode) { taosThreadMutexLock(&pBuf->mutex); + syncLogBufferValidate(pBuf); SyncIndex lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore); ASSERT(lastVer == pBuf->matchIndex); SyncIndex index = pBuf->endIndex - 1; @@ -1089,6 +1099,7 @@ int32_t syncLogBufferReset(SSyncLogBuffer* pBuf, SSyncNode* pNode) { SSyncLogReplMgr* pMgr = pNode->logReplMgrs[i]; syncLogReplMgrReset(pMgr); } + syncLogBufferValidate(pBuf); taosThreadMutexUnlock(&pBuf->mutex); return 0; } From 3b47dd753e4fb3c620582e278e13de9c3568c317 Mon Sep 17 00:00:00 2001 From: Benguang Zhao Date: Tue, 21 Feb 2023 20:01:54 +0800 Subject: [PATCH 15/34] enh: update pSyncNode->minMatchIndex in sncNodeEqPeerHeartbeatTimer --- source/libs/sync/src/syncMain.c | 16 +++------------- 1 file changed, 3 insertions(+), 13 deletions(-) diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 1deb9fa066..84341803d1 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -2094,24 +2094,19 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { if (timerLogicClock == msgLogicClock) { if (tsNow > pData->execTime) { -#if 0 - sTrace( - "vgId:%d, hbDataRid:%ld, EXECUTE this step-------- heartbeat tsNow:%ld, exec:%ld, tsNow-exec:%ld, " - "---------", - pSyncNode->vgId, hbDataRid, tsNow, pData->execTime, tsNow - pData->execTime); -#endif - pData->execTime += pSyncTimer->timerMS; SRpcMsg rpcMsg = {0}; (void)syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId); + pSyncNode->minMatchIndex = syncMinMatchIndex(pSyncNode); + SyncHeartbeat* pSyncMsg = rpcMsg.pCont; pSyncMsg->srcId = pSyncNode->myRaftId; pSyncMsg->destId = pData->destId; pSyncMsg->term = raftStoreGetTerm(pSyncNode); pSyncMsg->commitIndex = pSyncNode->commitIndex; - pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode); + pSyncMsg->minMatchIndex = pSyncNode->minMatchIndex; pSyncMsg->privateTerm = 0; pSyncMsg->timeStamp = tsNow; @@ -2123,11 +2118,6 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { syncLogSendHeartbeat(pSyncNode, pSyncMsg, false, timerElapsed, pData->execTime); syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg); } else { -#if 0 - sTrace( - "vgId:%d, hbDataRid:%ld, pass this step-------- heartbeat tsNow:%ld, exec:%ld, tsNow-exec:%ld, ---------", - pSyncNode->vgId, hbDataRid, tsNow, pData->execTime, tsNow - pData->execTime); -#endif } if (syncIsInit()) { From 644ba11a55bbf9bfacbf7a6c8d75256d03d7a1ed Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Wed, 22 Feb 2023 18:46:59 +0800 Subject: [PATCH 16/34] fix: fix asan error --- source/libs/scalar/src/filter.c | 4 ++++ tests/script/sh/checkAsan.sh | 5 ++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/source/libs/scalar/src/filter.c b/source/libs/scalar/src/filter.c index 25e65d2588..d4444ead0f 100644 --- a/source/libs/scalar/src/filter.c +++ b/source/libs/scalar/src/filter.c @@ -3146,6 +3146,10 @@ static FORCE_INLINE bool filterExecuteImplNotNull(void *pinfo, int32_t numOfRows for (int32_t i = 0; i < numOfRows; ++i) { uint32_t uidx = info->groups[0].unitIdxs[0]; + if (((SColumnInfoData *)info->cunits[uidx].colData)->pData == NULL) { + continue; + } + void *colData = colDataGetData((SColumnInfoData *)info->cunits[uidx].colData, i); p[i] = ((colData != NULL) && !colDataIsNull((SColumnInfoData *)info->cunits[uidx].colData, 0, i, NULL)); diff --git a/tests/script/sh/checkAsan.sh b/tests/script/sh/checkAsan.sh index 5c7976a9fc..00702d95ed 100755 --- a/tests/script/sh/checkAsan.sh +++ b/tests/script/sh/checkAsan.sh @@ -40,7 +40,10 @@ python_error=`cat ${LOG_DIR}/*.info | grep -w "stack" | wc -l` # /root/TDengine/source/common/src/tdataformat.c:1876:7: runtime error: signed integer overflow: 8252423483843671206 + 2406154664059062870 cannot be represented in type 'long int' # /home/chr/TDengine/source/libs/scalar/src/filter.c:3149:14: runtime error: applying non-zero offset 18446744073709551615 to null pointer -runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | grep -v "trees.c:873" | grep -v "sclfunc.c.*outside the range of representable values of type"| grep -v "signed integer overflow" |grep -v "strerror.c"| grep -v "asan_malloc_linux.cc" |grep -v "filter.c:3149:14"|wc -l` +# /home/chr/TDengine/source/libs/scalar/src/filter.c:3149:14: runtime error: applying non-zero offset 18446744073709551615 to null pointer +# /home/TDinternal/community/source/libs/scalar/src/sclvector.c:1109:66: runtime error: signed integer overflow: 9223372034707292160 + 1676867897049 cannot be represented in type 'long int' + +runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | grep -v "trees.c:873" | grep -v "sclfunc.c.*outside the range of representable values of type"| grep -v "signed integer overflow" |grep -v "strerror.c"| grep -v "asan_malloc_linux.cc" |wc -l` echo -e "\033[44;32;1m"asan error_num: $error_num"\033[0m" echo -e "\033[44;32;1m"asan memory_leak: $memory_leak"\033[0m" From ecaf5193a2f2625ce5c2665536fd9090dfbd803a Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Wed, 22 Feb 2023 18:46:59 +0800 Subject: [PATCH 17/34] fix: fix asan error --- tests/script/sh/checkAsan.sh | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/script/sh/checkAsan.sh b/tests/script/sh/checkAsan.sh index 00702d95ed..2bd4eaa548 100755 --- a/tests/script/sh/checkAsan.sh +++ b/tests/script/sh/checkAsan.sh @@ -40,9 +40,6 @@ python_error=`cat ${LOG_DIR}/*.info | grep -w "stack" | wc -l` # /root/TDengine/source/common/src/tdataformat.c:1876:7: runtime error: signed integer overflow: 8252423483843671206 + 2406154664059062870 cannot be represented in type 'long int' # /home/chr/TDengine/source/libs/scalar/src/filter.c:3149:14: runtime error: applying non-zero offset 18446744073709551615 to null pointer -# /home/chr/TDengine/source/libs/scalar/src/filter.c:3149:14: runtime error: applying non-zero offset 18446744073709551615 to null pointer -# /home/TDinternal/community/source/libs/scalar/src/sclvector.c:1109:66: runtime error: signed integer overflow: 9223372034707292160 + 1676867897049 cannot be represented in type 'long int' - runtime_error=`cat ${LOG_DIR}/*.asan | grep "runtime error" | grep -v "trees.c:873" | grep -v "sclfunc.c.*outside the range of representable values of type"| grep -v "signed integer overflow" |grep -v "strerror.c"| grep -v "asan_malloc_linux.cc" |wc -l` echo -e "\033[44;32;1m"asan error_num: $error_num"\033[0m" From 78d72a7db262dbdea6d30156bae229bd59c6c35c Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Thu, 23 Feb 2023 14:21:14 +0800 Subject: [PATCH 18/34] fix: alter column length too big issue --- source/libs/parser/src/parTranslater.c | 22 ++++++++++++++++------ source/util/src/terror.c | 2 +- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 0d052846f7..8b3cfd105b 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -5012,14 +5012,24 @@ static int32_t checkAlterSuperTableBySchema(STranslateContext* pCxt, SAlterTable return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_MODIFY_COL); } - if (TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES == pStmt->alterType && - pTableMeta->tableInfo.rowSize + calcTypeBytes(pStmt->dataType) - pSchema->bytes > TSDB_MAX_BYTES_PER_ROW) { - return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ROW_LENGTH, TSDB_MAX_BYTES_PER_ROW); + if (TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES == pStmt->alterType) { + if (calcTypeBytes(pStmt->dataType) > TSDB_MAX_FIELD_LEN) { + return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN); + } + + if (pTableMeta->tableInfo.rowSize + calcTypeBytes(pStmt->dataType) - pSchema->bytes > TSDB_MAX_BYTES_PER_ROW) { + return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ROW_LENGTH, TSDB_MAX_BYTES_PER_ROW); + } } - if (TSDB_ALTER_TABLE_UPDATE_TAG_BYTES == pStmt->alterType && - tagsLen + calcTypeBytes(pStmt->dataType) - pSchema->bytes > TSDB_MAX_TAGS_LEN) { - return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TAGS_LENGTH, TSDB_MAX_TAGS_LEN); + if (TSDB_ALTER_TABLE_UPDATE_TAG_BYTES == pStmt->alterType) { + if (calcTypeBytes(pStmt->dataType) > TSDB_MAX_FIELD_LEN) { + return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN); + } + + if (tagsLen + calcTypeBytes(pStmt->dataType) - pSchema->bytes > TSDB_MAX_TAGS_LEN) { + return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TAGS_LENGTH, TSDB_MAX_TAGS_LEN); + } } } diff --git a/source/util/src/terror.c b/source/util/src/terror.c index c07fa88af5..b85035ffcf 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -514,7 +514,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_ROW_LENGTH, "Row length exceeds TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_COLUMNS_NUM, "Illegal number of columns") TAOS_DEFINE_ERROR(TSDB_CODE_PAR_TOO_MANY_COLUMNS, "Too many columns") TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_FIRST_COLUMN, "First column must be timestamp") -TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN, "Invalid binary/nchar column length") +TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN, "Invalid binary/nchar column/tag length") TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_TAGS_NUM, "Invalid number of tag columns") TAOS_DEFINE_ERROR(TSDB_CODE_PAR_PERMISSION_DENIED, "Permission denied") TAOS_DEFINE_ERROR(TSDB_CODE_PAR_INVALID_STREAM_QUERY, "Invalid stream query") From 72e7f617e258806760bd776c8536aee490e53303 Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao@163.com> Date: Thu, 23 Feb 2023 14:24:45 +0800 Subject: [PATCH 19/34] fix:disable stream --- source/dnode/vnode/src/tq/tq.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 0cabce435a..ba34373282 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1060,6 +1060,9 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msg code = streamMetaAddSerializedTask(pTq->pStreamMeta, version, msg, msgLen); if (code < 0) return code; #endif + if (tsDisableStream) { + return 0; + } // 1.deserialize msg and build task SStreamTask* pTask = taosMemoryCalloc(1, sizeof(SStreamTask)); From 25c555a50226cdd751d2f57f541b9adbf7ab8289 Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Thu, 23 Feb 2023 14:38:57 +0800 Subject: [PATCH 20/34] fix: refresh sys db vgroups after creating database --- source/client/src/clientMsgHandler.c | 16 ++++++++++++++++ tests/script/tsim/catalog/alterInCurrent.sim | 15 +++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/source/client/src/clientMsgHandler.c b/source/client/src/clientMsgHandler.c index f414c7e92f..ed54144858 100644 --- a/source/client/src/clientMsgHandler.c +++ b/source/client/src/clientMsgHandler.c @@ -163,6 +163,22 @@ int32_t processCreateDbRsp(void* param, SDataBuf* pMsg, int32_t code) { taosMemoryFree(pMsg->pEpSet); if (code != TSDB_CODE_SUCCESS) { setErrno(pRequest, code); + } else { + struct SCatalog* pCatalog = NULL; + int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); + if (TSDB_CODE_SUCCESS == code) { + STscObj* pTscObj = pRequest->pTscObj; + + SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + char dbFName[TSDB_DB_FNAME_LEN]; + snprintf(dbFName, sizeof(dbFName) - 1, "%d.%s", pTscObj->acctId, TSDB_INFORMATION_SCHEMA_DB); + catalogRefreshDBVgInfo(pCatalog, &conn, dbFName); + snprintf(dbFName, sizeof(dbFName) - 1, "%d.%s", pTscObj->acctId, TSDB_PERFORMANCE_SCHEMA_DB); + catalogRefreshDBVgInfo(pCatalog, &conn, dbFName); + } } if (pRequest->body.queryFp) { diff --git a/tests/script/tsim/catalog/alterInCurrent.sim b/tests/script/tsim/catalog/alterInCurrent.sim index 3cb337bbe1..521858c368 100644 --- a/tests/script/tsim/catalog/alterInCurrent.sim +++ b/tests/script/tsim/catalog/alterInCurrent.sim @@ -67,4 +67,19 @@ sql insert into t1 values (1591060628000, 1); sql alter table st1 drop tag t2; sql create table t2 using st1 tags(2); +print ======== drop tag in super table +sql create database if not exists aaa; +sql select table_name, db_name from information_schema.ins_tables t where t.db_name like 'aaa'; +if $rows != 0 then + return -1 +endi +sql drop database if exists foo; +sql create database if not exists foo; +sql create table foo.t(ts timestamp,name varchar(20)); +sql create table foo.xt(ts timestamp,name varchar(20)); +sql select table_name, db_name from information_schema.ins_tables t where t.db_name like 'foo'; +if $rows != 2 then + return -1 +endi + system sh/exec.sh -n dnode1 -s stop -x SIGINT From 8ea6b545b8037a0f9f77f51886811cebb1983c57 Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Thu, 23 Feb 2023 15:04:31 +0800 Subject: [PATCH 21/34] fix: restore catalog job error --- source/libs/catalog/src/ctgAsync.c | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c index 1276f2796f..89e92b0cc8 100644 --- a/source/libs/catalog/src/ctgAsync.c +++ b/source/libs/catalog/src/ctgAsync.c @@ -906,14 +906,9 @@ int32_t ctgCallUserCb(void* param) { } void ctgUpdateJobErrCode(SCtgJob* pJob, int32_t errCode) { - if (errCode == TSDB_CODE_SUCCESS) return; + if (!NEED_CLIENT_REFRESH_VG_ERROR(errCode) || errCode == TSDB_CODE_SUCCESS) return; - if (NEED_CLIENT_HANDLE_ERROR(errCode)) { - atomic_store_32(&pJob->jobResCode, errCode); - } else if (0 != atomic_val_compare_exchange_32(&pJob->jobResCode, 0, errCode)) { - return; - } - + atomic_store_32(&pJob->jobResCode, errCode); qDebug("QID:0x%" PRIx64 " ctg job errCode updated to %s", pJob->queryId, tstrerror(errCode)); return; } From 797722a6ae0fcbec60ee418b8442712f999b1dc9 Mon Sep 17 00:00:00 2001 From: Minglei Jin Date: Thu, 23 Feb 2023 15:43:34 +0800 Subject: [PATCH 22/34] fix(tdb/coder): free key memory of stack btc's coder --- source/libs/tdb/src/db/tdbBtree.c | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c index a35e01c7aa..6353d8ba27 100644 --- a/source/libs/tdb/src/db/tdbBtree.c +++ b/source/libs/tdb/src/db/tdbBtree.c @@ -264,7 +264,10 @@ int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, i // move the cursor ret = tdbBtcMoveTo(&btc, pKey, nKey, &c); if (ret < 0) { - ASSERT(0); + tdbError("tdb/btree-upsert: btc move to failed with ret: %d.", ret); + if (TDB_CELLDECODER_FREE_KEY(&btc.coder)) { + tdbFree(btc.coder.pKey); + } tdbBtcClose(&btc); return -1; } @@ -280,11 +283,17 @@ int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, i ret = tdbBtcUpsert(&btc, pKey, nKey, pData, nData, c); if (ret < 0) { - ASSERT(0); + if (TDB_CELLDECODER_FREE_KEY(&btc.coder)) { + tdbFree(btc.coder.pKey); + } tdbBtcClose(&btc); + tdbError("tdb/btree-upsert: btc upsert failed with ret: %d.", ret); return -1; } + if (TDB_CELLDECODER_FREE_KEY(&btc.coder)) { + tdbFree(btc.coder.pKey); + } tdbBtcClose(&btc); return 0; } @@ -2188,10 +2197,6 @@ int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) { } else { lidx = lidx + 1; } - if (TDB_CELLDECODER_FREE_KEY(&pBtc->coder)) { - tdbFree((void*)pTKey); - } - // compare last cell if (lidx <= ridx) { pBtc->idx = ridx; @@ -2202,9 +2207,6 @@ int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) { } else { ridx = ridx - 1; } - if (TDB_CELLDECODER_FREE_KEY(&pBtc->coder)) { - tdbFree((void*)pTKey); - } } // binary search @@ -2215,9 +2217,6 @@ int tdbBtcMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) { pBtc->idx = (lidx + ridx) >> 1; tdbBtcGet(pBtc, &pTKey, &tkLen, NULL, NULL); c = pBt->kcmpr(pKey, kLen, pTKey, tkLen); - if (TDB_CELLDECODER_FREE_KEY(&pBtc->coder)) { - tdbFree((void*)pTKey); - } if (c < 0) { // pKey < cd.pKey ridx = pBtc->idx - 1; From cda49b559c9abf0c1939dda7b4643ad6554d5f66 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Wed, 22 Feb 2023 18:46:59 +0800 Subject: [PATCH 23/34] fix: fix asan error --- source/libs/scalar/src/filter.c | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/source/libs/scalar/src/filter.c b/source/libs/scalar/src/filter.c index d4444ead0f..d91b2ebd6b 100644 --- a/source/libs/scalar/src/filter.c +++ b/source/libs/scalar/src/filter.c @@ -3120,9 +3120,8 @@ static FORCE_INLINE bool filterExecuteImplIsNull(void *pinfo, int32_t numOfRows, for (int32_t i = 0; i < numOfRows; ++i) { uint32_t uidx = info->groups[0].unitIdxs[0]; - void *colData = colDataGetData((SColumnInfoData *)info->cunits[uidx].colData, i); - p[i] = ((colData == NULL) || colDataIsNull((SColumnInfoData *)info->cunits[uidx].colData, 0, i, NULL)); + p[i] = colDataIsNull((SColumnInfoData *)info->cunits[uidx].colData, 0, i, NULL); if (p[i] == 0) { all = false; } else { @@ -3146,13 +3145,8 @@ static FORCE_INLINE bool filterExecuteImplNotNull(void *pinfo, int32_t numOfRows for (int32_t i = 0; i < numOfRows; ++i) { uint32_t uidx = info->groups[0].unitIdxs[0]; - if (((SColumnInfoData *)info->cunits[uidx].colData)->pData == NULL) { - continue; - } - void *colData = colDataGetData((SColumnInfoData *)info->cunits[uidx].colData, i); - - p[i] = ((colData != NULL) && !colDataIsNull((SColumnInfoData *)info->cunits[uidx].colData, 0, i, NULL)); + p[i] = !colDataIsNull((SColumnInfoData *)info->cunits[uidx].colData, 0, i, NULL); if (p[i] == 0) { all = false; } else { @@ -3182,13 +3176,13 @@ bool filterExecuteImplRange(void *pinfo, int32_t numOfRows, SColumnInfoData *pRe for (int32_t i = 0; i < numOfRows; ++i) { SColumnInfoData *pData = info->cunits[0].colData; - void *colData = colDataGetData(pData, i); - if (colData == NULL || colDataIsNull_s(pData, i)) { + if (colDataIsNull_s(pData, i)) { all = false; p[i] = 0; continue; } + void *colData = colDataGetData(pData, i); p[i] = (*rfunc)(colData, colData, valData, valData2, func); if (p[i] == 0) { @@ -3214,13 +3208,14 @@ bool filterExecuteImplMisc(void *pinfo, int32_t numOfRows, SColumnInfoData *pRes for (int32_t i = 0; i < numOfRows; ++i) { uint32_t uidx = info->groups[0].unitIdxs[0]; - void *colData = colDataGetData((SColumnInfoData *)info->cunits[uidx].colData, i); - if (colData == NULL || colDataIsNull_s((SColumnInfoData *)info->cunits[uidx].colData, i)) { + + if (colDataIsNull_s((SColumnInfoData *)info->cunits[uidx].colData, i)) { p[i] = 0; all = false; continue; } + void *colData = colDataGetData((SColumnInfoData *)info->cunits[uidx].colData, i); // match/nmatch for nchar type need convert from ucs4 to mbs if (info->cunits[uidx].dataType == TSDB_DATA_TYPE_NCHAR && (info->cunits[uidx].optr == OP_TYPE_MATCH || info->cunits[uidx].optr == OP_TYPE_NMATCH)) { @@ -3278,7 +3273,7 @@ bool filterExecuteImpl(void *pinfo, int32_t numOfRows, SColumnInfoData *pRes, SC if (!isNull) { colData = colDataGetData((SColumnInfoData *)(cunit->colData), i); } - + if (colData == NULL || isNull) { p[i] = optr == OP_TYPE_IS_NULL ? true : false; } else { From 0833592f832bcccaec12ea97d067324323814d17 Mon Sep 17 00:00:00 2001 From: Minglei Jin Date: Thu, 23 Feb 2023 19:42:03 +0800 Subject: [PATCH 24/34] fix(tdb): free realloced coder's pKey --- source/libs/tdb/src/db/tdbBtree.c | 22 ++++++++++++---------- source/libs/tdb/src/inc/tdbInt.h | 2 ++ 2 files changed, 14 insertions(+), 10 deletions(-) diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c index 6353d8ba27..bf8c5c53dc 100644 --- a/source/libs/tdb/src/db/tdbBtree.c +++ b/source/libs/tdb/src/db/tdbBtree.c @@ -253,7 +253,7 @@ int tdbBtreeDelete(SBTree *pBt, const void *pKey, int kLen, TXN *pTxn) { } int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, int nData, TXN *pTxn) { - SBTC btc; + SBTC btc = {0}; int c; int ret; @@ -272,6 +272,10 @@ int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, i return -1; } + if (TDB_CELLDECODER_FREE_KEY(&btc.coder)) { + tdbFree(btc.coder.pKey); + } + if (btc.idx == -1) { btc.idx = 0; c = 1; @@ -283,17 +287,11 @@ int tdbBtreeUpsert(SBTree *pBt, const void *pKey, int nKey, const void *pData, i ret = tdbBtcUpsert(&btc, pKey, nKey, pData, nData, c); if (ret < 0) { - if (TDB_CELLDECODER_FREE_KEY(&btc.coder)) { - tdbFree(btc.coder.pKey); - } tdbBtcClose(&btc); tdbError("tdb/btree-upsert: btc upsert failed with ret: %d.", ret); return -1; } - if (TDB_CELLDECODER_FREE_KEY(&btc.coder)) { - tdbFree(btc.coder.pKey); - } tdbBtcClose(&btc); return 0; } @@ -1437,15 +1435,19 @@ static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pD // Clear the state of decoder if (TDB_CELLDECODER_FREE_VAL(pDecoder)) { tdbFree(pDecoder->pVal); + TDB_CELLDECODER_CLZ_FREE_VAL(pDecoder); + // tdbTrace("tdb btc decoder val set nil: %p/0x%x ", pDecoder, pDecoder->freeKV); + } + if (TDB_CELLDECODER_FREE_KEY(pDecoder)) { + tdbFree(pDecoder->pKey); + TDB_CELLDECODER_CLZ_FREE_KEY(pDecoder); + // tdbTrace("tdb btc decoder key set nil: %p/0x%x ", pDecoder, pDecoder->freeKV); } pDecoder->kLen = -1; pDecoder->pKey = NULL; pDecoder->vLen = -1; pDecoder->pVal = NULL; pDecoder->pgno = 0; - TDB_CELLDECODER_SET_FREE_NIL(pDecoder); - - // tdbTrace("tdb btc decoder set nil: %p/0x%x ", pDecoder, pDecoder->freeKV); // 1. Decode header part if (!leaf) { diff --git a/source/libs/tdb/src/inc/tdbInt.h b/source/libs/tdb/src/inc/tdbInt.h index 62466e9c47..7a0bcc00a4 100644 --- a/source/libs/tdb/src/inc/tdbInt.h +++ b/source/libs/tdb/src/inc/tdbInt.h @@ -122,6 +122,8 @@ typedef struct SBtInfo { #define TDB_CELLD_F_VAL 0x2 #define TDB_CELLDECODER_SET_FREE_NIL(pCellDecoder) ((pCellDecoder)->freeKV = TDB_CELLD_F_NIL) +#define TDB_CELLDECODER_CLZ_FREE_KEY(pCellDecoder) ((pCellDecoder)->freeKV &= ~TDB_CELLD_F_KEY) +#define TDB_CELLDECODER_CLZ_FREE_VAL(pCellDecoder) ((pCellDecoder)->freeKV &= ~TDB_CELLD_F_VAL) #define TDB_CELLDECODER_SET_FREE_KEY(pCellDecoder) ((pCellDecoder)->freeKV |= TDB_CELLD_F_KEY) #define TDB_CELLDECODER_SET_FREE_VAL(pCellDecoder) ((pCellDecoder)->freeKV |= TDB_CELLD_F_VAL) From f477c4832e4aa3b63a6cadfe59babbe231317ecb Mon Sep 17 00:00:00 2001 From: xiaolei li <85657333+xleili@users.noreply.github.com> Date: Thu, 23 Feb 2023 21:13:13 +0800 Subject: [PATCH 25/34] release: update version to 3.0.2.6 (#20135) --- cmake/cmake.version | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmake/cmake.version b/cmake/cmake.version index a30618157b..d0d455c73d 100644 --- a/cmake/cmake.version +++ b/cmake/cmake.version @@ -2,7 +2,7 @@ IF (DEFINED VERNUMBER) SET(TD_VER_NUMBER ${VERNUMBER}) ELSE () - SET(TD_VER_NUMBER "3.0.2.5") + SET(TD_VER_NUMBER "3.0.2.6") ENDIF () IF (DEFINED VERCOMPATIBLE) From a73bd389854df8afe6f3162d101dd89919c34502 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 23 Feb 2023 22:05:09 +0800 Subject: [PATCH 26/34] fix: opt trans debug info --- source/libs/transport/src/transCli.c | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 5d6751a260..1ba385e3a2 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -1949,11 +1949,13 @@ static void cliSchedMsgToNextNode(SCliMsg* pMsg, SCliThrd* pThrd) { STrans* pTransInst = pThrd->pTransInst; STransConnCtx* pCtx = pMsg->ctx; - STraceId* trace = &pMsg->msg.info.traceId; - char tbuf[256] = {0}; - EPSET_DEBUG_STR(&pCtx->epSet, tbuf); - tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", transLabel(pThrd->pTransInst), tbuf, - pCtx->retryStep, pCtx->retryNextInterval); + if (rpcDebugFlag & DEBUG_DEBUG) { + STraceId* trace = &pMsg->msg.info.traceId; + char tbuf[256] = {0}; + EPSET_DEBUG_STR(&pCtx->epSet, tbuf); + tGDebug("%s retry on next node,use:%s, step: %d,timeout:%" PRId64 "", transLabel(pThrd->pTransInst), tbuf, + pCtx->retryStep, pCtx->retryNextInterval); + } STaskArg* arg = taosMemoryMalloc(sizeof(STaskArg)); arg->param1 = pMsg; @@ -2181,9 +2183,11 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { STraceId* trace = &pResp->info.traceId; bool hasEpSet = cliTryExtractEpSet(pResp, &pCtx->epSet); if (hasEpSet) { - char tbuf[256] = {0}; - EPSET_DEBUG_STR(&pCtx->epSet, tbuf); - tGTrace("%s conn %p extract epset from msg", CONN_GET_INST_LABEL(pConn), pConn); + if (rpcDebugFlag & DEBUG_TRACE) { + char tbuf[256] = {0}; + EPSET_DEBUG_STR(&pCtx->epSet, tbuf); + tGTrace("%s conn %p extract epset from msg", CONN_GET_INST_LABEL(pConn), pConn); + } } if (pCtx->pSem != NULL) { From 0d5aad85f878d54588300de17c2666ec01559bd2 Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Thu, 23 Feb 2023 22:29:56 +0800 Subject: [PATCH 27/34] fix: opt trans debug info --- source/libs/transport/inc/transComm.h | 1 + source/libs/transport/src/transCli.c | 20 ++++++++------------ 2 files changed, 9 insertions(+), 12 deletions(-) diff --git a/source/libs/transport/inc/transComm.h b/source/libs/transport/inc/transComm.h index a41cc0068c..5ff67c87ca 100644 --- a/source/libs/transport/inc/transComm.h +++ b/source/libs/transport/inc/transComm.h @@ -22,6 +22,7 @@ extern "C" { #include "os.h" #include "taoserror.h" #include "theap.h" +#include "tmisce.h" #include "transLog.h" #include "transportInt.h" #include "trpc.h" diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 1ba385e3a2..a755467ccc 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -11,7 +11,6 @@ * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ - #include "transComm.h" typedef struct SConnList { @@ -1992,7 +1991,7 @@ FORCE_INLINE bool cliTryExtractEpSet(STransMsg* pResp, SEpSet* dst) { pResp->pCont = buf; pResp->contLen = len; - *dst = epset; + epsetAssign(dst, &epset); return true; } bool cliResetEpset(STransConnCtx* pCtx, STransMsg* pResp, bool hasEpSet) { @@ -2017,7 +2016,7 @@ bool cliResetEpset(STransConnCtx* pCtx, STransMsg* pResp, bool hasEpSet) { } else { if (!transEpSetIsEqual(&pCtx->epSet, &epSet)) { tDebug("epset not equal, retry new epset"); - pCtx->epSet = epSet; + epsetAssign(&pCtx->epSet, &epSet); noDelay = false; } else { if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { @@ -2042,7 +2041,7 @@ bool cliResetEpset(STransConnCtx* pCtx, STransMsg* pResp, bool hasEpSet) { } else { if (!transEpSetIsEqual(&pCtx->epSet, &epSet)) { tDebug("epset not equal, retry new epset"); - pCtx->epSet = epSet; + epsetAssign(&pCtx->epSet, &epSet); noDelay = false; } else { if (pCtx->epsetRetryCnt >= pCtx->epSet.numOfEps) { @@ -2132,10 +2131,6 @@ bool cliGenRetryRule(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) { if (pCtx->retryNextInterval >= pCtx->retryMaxInterval) { pCtx->retryNextInterval = pCtx->retryMaxInterval; } - - // if (-1 != pCtx->retryMaxTimeout && taosGetTimestampMs() - pCtx->retryInitTimestamp >= pCtx->retryMaxTimeout) { - // return false; - // } } else { pCtx->retryNextInterval = 0; pCtx->epsetRetryCnt++; @@ -2314,8 +2309,9 @@ int transSendRequest(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STran TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64()); STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx)); - pCtx->epSet = *pEpSet; - pCtx->origEpSet = *pEpSet; + epsetAssign(&pCtx->epSet, pEpSet); + epsetAssign(&pCtx->origEpSet, pEpSet); + pCtx->ahandle = pReq->info.ahandle; pCtx->msgType = pReq->msgType; @@ -2360,8 +2356,8 @@ int transSendRecv(void* shandle, const SEpSet* pEpSet, STransMsg* pReq, STransMs TRACE_SET_MSGID(&pReq->info.traceId, tGenIdPI64()); STransConnCtx* pCtx = taosMemoryCalloc(1, sizeof(STransConnCtx)); - pCtx->epSet = *pEpSet; - pCtx->origEpSet = *pEpSet; + epsetAssign(&pCtx->epSet, pEpSet); + epsetAssign(&pCtx->origEpSet, pEpSet); pCtx->ahandle = pReq->info.ahandle; pCtx->msgType = pReq->msgType; pCtx->pSem = sem; From f6ca06ef2bc34af71d4651c34ad12a8201ffa65a Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Fri, 24 Feb 2023 09:47:38 +0800 Subject: [PATCH 28/34] fix: opt trans debug info --- source/libs/transport/src/transCli.c | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index a755467ccc..38189f90db 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -223,9 +223,13 @@ static void cliWalkCb(uv_handle_t* handle, void* arg); } while (0); // snprintf may cause performance problem -#define CONN_CONSTRUCT_HASH_KEY(key, ip, port) \ - do { \ - snprintf(key, sizeof(key), "%s:%d", ip, (int)port); \ +#define CONN_CONSTRUCT_HASH_KEY(key, ip, port) \ + do { \ + char* p = key; \ + int32_t len = strlen(ip); \ + if (p != NULL) memcpy(p, ip, len); \ + p[len] = ':'; \ + titoa(port, 10, &p[len + 1]); \ } while (0) #define CONN_PERSIST_TIME(para) ((para) <= 90000 ? 90000 : (para)) @@ -663,7 +667,7 @@ static int32_t specifyConnRef(SCliConn* conn, bool update, int64_t handle) { static void cliAllocRecvBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { SCliConn* conn = handle->data; SConnBuffer* pBuf = &conn->readBuf; - tDebug("%s conn %p alloc read buf", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p alloc read buf", CONN_GET_INST_LABEL(conn), conn); transAllocBuffer(pBuf, buf); } static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { @@ -676,7 +680,7 @@ static void cliRecvCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { if (nread > 0) { pBuf->len += nread; while (transReadComplete(pBuf)) { - tDebug("%s conn %p read complete", CONN_GET_INST_LABEL(conn), conn); + tTrace("%s conn %p read complete", CONN_GET_INST_LABEL(conn), conn); if (pBuf->invalid) { cliHandleExcept(conn); break; From d2b4b29c825d6cf9718c61697d39d1cb812aa388 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Fri, 24 Feb 2023 11:00:43 +0800 Subject: [PATCH 29/34] fix: taosbenchmark coverity scan issues for main (#20141) --- cmake/taostools_CMakeLists.txt.in | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmake/taostools_CMakeLists.txt.in b/cmake/taostools_CMakeLists.txt.in index ae3b626f88..db2ae92f6e 100644 --- a/cmake/taostools_CMakeLists.txt.in +++ b/cmake/taostools_CMakeLists.txt.in @@ -2,7 +2,7 @@ # taos-tools ExternalProject_Add(taos-tools GIT_REPOSITORY https://github.com/taosdata/taos-tools.git - GIT_TAG 61cbfd2 + GIT_TAG 1e15545 SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools" BINARY_DIR "" #BUILD_IN_SOURCE TRUE From 84cf3ec69660a8c5f521ccfcbefee5458b572c85 Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao@163.com> Date: Fri, 17 Feb 2023 14:20:45 +0800 Subject: [PATCH 30/34] fix:add log --- source/libs/executor/src/executor.c | 2 +- source/libs/executor/src/groupoperator.c | 5 +---- source/libs/executor/src/scanoperator.c | 2 +- 3 files changed, 3 insertions(+), 6 deletions(-) diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 366ef09e60..e5e9b45576 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -107,7 +107,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu pOperator->status = OP_NOT_OPENED; SStreamScanInfo* pInfo = pOperator->info; - +///// numOfBlocks总和 日志 ASSERT(pInfo->validBlockIndex == 0); ASSERT(taosArrayGetSize(pInfo->pBlockLists) == 0); diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index 161fe52367..aa61d24b92 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -1064,11 +1064,8 @@ static SSDataBlock* doStreamHashPartition(SOperatorInfo* pOperator) { // there is an scalar expression that needs to be calculated right before apply the group aggregation. if (pInfo->scalarSup.pExprInfo != NULL) { - pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx, + projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx, pInfo->scalarSup.numOfExprs, NULL); - if (pTaskInfo->code != TSDB_CODE_SUCCESS) { - longjmp(pTaskInfo->env, pTaskInfo->code); - } } taosHashClear(pInfo->pPartitions); doStreamHashPartitionImpl(pInfo, pBlock); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 71e1068fb3..05f26332f2 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1798,7 +1798,7 @@ FETCH_NEXT_BLOCK: /*pOperator->status = OP_EXEC_DONE;*/ return NULL; } - +//////todo int32_t current = pInfo->validBlockIndex++; SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current); if (pBlock->info.id.groupId && pBlock->info.parTbName[0]) { From 40f9aaa7153327f05fb8096d888bd2e10ba3ac5e Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao@163.com> Date: Fri, 24 Feb 2023 11:00:08 +0800 Subject: [PATCH 31/34] fix:Ignore expression exception & add some log --- source/libs/executor/src/executor.c | 2 +- source/libs/executor/src/filloperator.c | 1 + source/libs/executor/src/scanoperator.c | 2 +- source/libs/executor/src/timewindowoperator.c | 6 +++--- 4 files changed, 6 insertions(+), 5 deletions(-) diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index e5e9b45576..82f079e2fb 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -107,7 +107,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu pOperator->status = OP_NOT_OPENED; SStreamScanInfo* pInfo = pOperator->info; -///// numOfBlocks总和 日志 + qDebug("stream set total blocks:%d, task id:%s" PRIx64, (int32_t)numOfBlocks, id); ASSERT(pInfo->validBlockIndex == 0); ASSERT(taosArrayGetSize(pInfo->pBlockLists) == 0); diff --git a/source/libs/executor/src/filloperator.c b/source/libs/executor/src/filloperator.c index 091b10a63e..a1426e2a96 100644 --- a/source/libs/executor/src/filloperator.c +++ b/source/libs/executor/src/filloperator.c @@ -924,6 +924,7 @@ static void doStreamFillLinear(SStreamFillSupporter* pFillSup, SStreamFillInfo* static void keepResultInDiscBuf(SOperatorInfo* pOperator, uint64_t groupId, SResultRowData* pRow, int32_t len) { SWinKey key = {.groupId = groupId, .ts = pRow->key}; int32_t code = streamStateFillPut(pOperator->pTaskInfo->streamInfo.pState, &key, pRow->pRowVal, len); + qDebug("===stream===fill operator save key ts:%" PRId64 " group id:%" PRIu64 " code:%d", key.ts, key.groupId, code); ASSERT(code == TSDB_CODE_SUCCESS); } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 05f26332f2..71e1068fb3 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1798,7 +1798,7 @@ FETCH_NEXT_BLOCK: /*pOperator->status = OP_EXEC_DONE;*/ return NULL; } -//////todo + int32_t current = pInfo->validBlockIndex++; SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current); if (pBlock->info.id.groupId && pBlock->info.parTbName[0]) { diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 54be30028e..62d68d5ca2 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1538,16 +1538,16 @@ static void deleteIntervalDiscBuf(SStreamState* pState, SHashObj* pPullDataMap, code = streamStateGetKVByCur(pCur, &tmpKey, NULL, 0); if (code == TSDB_CODE_SUCCESS) { STimeWindow tw = getFinalTimeWindow(tmpKey.ts, pInterval); - qDebug("===stream===error stream state first key:%" PRId64 "-%" PRId64 ",%" PRId64 ",mark %" PRId64, tw.skey, + qDebug("===stream===error stream state first key:%" PRId64 "-%" PRId64 ",%" PRIu64 ",mark %" PRId64, tw.skey, tw.ekey, tmpKey.groupId, mark); } else { STimeWindow tw = getFinalTimeWindow(key->ts, pInterval); - qDebug("===stream===stream state first key:%" PRId64 "-%" PRId64 ",%" PRId64 ",mark %" PRId64, tw.skey, tw.ekey, + qDebug("===stream===stream state first key:%" PRId64 "-%" PRId64 ",%" PRIu64 ",mark %" PRId64, tw.skey, tw.ekey, key->groupId, mark); } } else { STimeWindow tw = getFinalTimeWindow(key->ts, pInterval); - qDebug("===stream===stream state first key:%" PRId64 "-%" PRId64 ",%" PRId64 ",mark %" PRId64, tw.skey, tw.ekey, + qDebug("===stream===stream state first key:%" PRId64 "-%" PRId64 ",%" PRIu64 ",mark %" PRId64, tw.skey, tw.ekey, key->groupId, mark); } streamStateFreeCur(pCur); From f6acf035c991d76eded48d3a0f6e7ff66c389417 Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Fri, 24 Feb 2023 11:37:16 +0800 Subject: [PATCH 32/34] fix: sys db vgroup update issue --- include/libs/qcom/query.h | 8 ++- source/client/src/clientHb.c | 84 +++++++++++++++++++--------- tests/system-test/7-tmq/tmq_taosx.py | 1 + 3 files changed, 63 insertions(+), 30 deletions(-) diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h index 5b640dce92..f2f7ac5699 100644 --- a/include/libs/qcom/query.h +++ b/include/libs/qcom/query.h @@ -26,6 +26,7 @@ extern "C" { #include "tlog.h" #include "tmsg.h" #include "tmsgcb.h" +#include "systable.h" typedef enum { JOB_TASK_STATUS_NULL = 0, @@ -284,9 +285,10 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t #define REQUEST_TOTAL_EXEC_TIMES 2 -#define IS_SYS_DBNAME(_dbname) \ - (((*(_dbname) == 'i') && (0 == strcmp(_dbname, TSDB_INFORMATION_SCHEMA_DB))) || \ - ((*(_dbname) == 'p') && (0 == strcmp(_dbname, TSDB_PERFORMANCE_SCHEMA_DB)))) +#define IS_INFORMATION_SCHEMA_DB(_name) ((*(_name) == 'i') && (0 == strcmp(_name, TSDB_INFORMATION_SCHEMA_DB))) +#define IS_PERFORMANCE_SCHEMA_DB(_name) ((*(_name) == 'p') && (0 == strcmp(_name, TSDB_PERFORMANCE_SCHEMA_DB))) + +#define IS_SYS_DBNAME(_dbname) (IS_INFORMATION_SCHEMA_DB(_dbname) || IS_PERFORMANCE_SCHEMA_DB(_dbname)) #define qFatal(...) \ do { \ diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index b01a871702..5c4bcf7946 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -49,6 +49,48 @@ static int32_t hbProcessUserAuthInfoRsp(void *value, int32_t valueLen, struct SC return TSDB_CODE_SUCCESS; } +static int32_t hbGenerateVgInfoFromRsp(SDBVgInfo **pInfo, SUseDbRsp *rsp) { + int32_t code = 0; + SDBVgInfo *vgInfo = taosMemoryCalloc(1, sizeof(SDBVgInfo)); + if (NULL == vgInfo) { + code = TSDB_CODE_OUT_OF_MEMORY; + return code; + } + + vgInfo->vgVersion = rsp->vgVersion; + vgInfo->stateTs = rsp->stateTs; + vgInfo->hashMethod = rsp->hashMethod; + vgInfo->hashPrefix = rsp->hashPrefix; + vgInfo->hashSuffix = rsp->hashSuffix; + vgInfo->vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK); + if (NULL == vgInfo->vgHash) { + taosMemoryFree(vgInfo); + tscError("hash init[%d] failed", rsp->vgNum); + code = TSDB_CODE_OUT_OF_MEMORY; + goto _return; + } + + for (int32_t j = 0; j < rsp->vgNum; ++j) { + SVgroupInfo *pInfo = taosArrayGet(rsp->pVgroupInfos, j); + if (taosHashPut(vgInfo->vgHash, &pInfo->vgId, sizeof(int32_t), pInfo, sizeof(SVgroupInfo)) != 0) { + tscError("hash push failed, errno:%d", errno); + taosHashCleanup(vgInfo->vgHash); + taosMemoryFree(vgInfo); + code = TSDB_CODE_OUT_OF_MEMORY; + goto _return; + } + } + +_return: + if (code) { + taosHashCleanup(vgInfo->vgHash); + taosMemoryFreeClear(vgInfo); + } + + *pInfo = vgInfo; + return code; +} + static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) { int32_t code = 0; @@ -67,37 +109,22 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog if (rsp->vgVersion < 0) { code = catalogRemoveDB(pCatalog, rsp->db, rsp->uid); } else { - SDBVgInfo *vgInfo = taosMemoryCalloc(1, sizeof(SDBVgInfo)); - if (NULL == vgInfo) { - code = TSDB_CODE_OUT_OF_MEMORY; + SDBVgInfo *vgInfo = NULL; + code = hbGenerateVgInfoFromRsp(&vgInfo, rsp); + if (TSDB_CODE_SUCCESS != code) { goto _return; } - vgInfo->vgVersion = rsp->vgVersion; - vgInfo->stateTs = rsp->stateTs; - vgInfo->hashMethod = rsp->hashMethod; - vgInfo->hashPrefix = rsp->hashPrefix; - vgInfo->hashSuffix = rsp->hashSuffix; - vgInfo->vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK); - if (NULL == vgInfo->vgHash) { - taosMemoryFree(vgInfo); - tscError("hash init[%d] failed", rsp->vgNum); - code = TSDB_CODE_OUT_OF_MEMORY; - goto _return; - } - - for (int32_t j = 0; j < rsp->vgNum; ++j) { - SVgroupInfo *pInfo = taosArrayGet(rsp->pVgroupInfos, j); - if (taosHashPut(vgInfo->vgHash, &pInfo->vgId, sizeof(int32_t), pInfo, sizeof(SVgroupInfo)) != 0) { - tscError("hash push failed, errno:%d", errno); - taosHashCleanup(vgInfo->vgHash); - taosMemoryFree(vgInfo); - code = TSDB_CODE_OUT_OF_MEMORY; - goto _return; - } - } - catalogUpdateDBVgInfo(pCatalog, rsp->db, rsp->uid, vgInfo); + + if (IS_SYS_DBNAME(rsp->db)) { + code = hbGenerateVgInfoFromRsp(&vgInfo, rsp); + if (TSDB_CODE_SUCCESS != code) { + goto _return; + } + + catalogUpdateDBVgInfo(pCatalog, (rsp->db[0] == 'i') ? TSDB_PERFORMANCE_SCHEMA_DB : TSDB_INFORMATION_SCHEMA_DB, rsp->uid, vgInfo); + } } if (code) { @@ -492,6 +519,9 @@ int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SCl for (int32_t i = 0; i < dbNum; ++i) { SDbVgVersion *db = &dbs[i]; + tscDebug("the %dth expired dbFName:%s, dbId:%" PRId64 ", vgVersion:%d, numOfTable:%d, startTs:%" PRId64, + i, db->dbFName, db->dbId, db->vgVersion, db->numOfTable, db->stateTs); + db->dbId = htobe64(db->dbId); db->vgVersion = htonl(db->vgVersion); db->numOfTable = htonl(db->numOfTable); diff --git a/tests/system-test/7-tmq/tmq_taosx.py b/tests/system-test/7-tmq/tmq_taosx.py index 0596241ce1..54bfab1ebc 100644 --- a/tests/system-test/7-tmq/tmq_taosx.py +++ b/tests/system-test/7-tmq/tmq_taosx.py @@ -195,6 +195,7 @@ class TDTestCase: tdSql.checkData(1, 1, 1) tdSql.checkData(1, 2, '{"k1":1,"k2":"hello"}') + time.sleep(10) tdSql.query("select * from information_schema.ins_tables where table_name = 'stt4'") uid1 = tdSql.getData(0, 5) uid2 = tdSql.getData(1, 5) From f6827c3f060242c809233de3d66881a0e7389cd7 Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Fri, 24 Feb 2023 13:53:48 +0800 Subject: [PATCH 33/34] fix: systable header file compile issue --- include/common/systable.h | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/include/common/systable.h b/include/common/systable.h index 6f65c1e8b8..cfc0af0172 100644 --- a/include/common/systable.h +++ b/include/common/systable.h @@ -12,6 +12,9 @@ * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ + +#ifndef TDENGINE_SYSTABLE_H +#define TDENGINE_SYSTABLE_H #ifdef __cplusplus extern "C" { @@ -19,9 +22,6 @@ extern "C" { #include "os.h" -#ifndef TDENGINE_SYSTABLE_H -#define TDENGINE_SYSTABLE_H - #define TSDB_INFORMATION_SCHEMA_DB "information_schema" #define TSDB_INS_TABLE_DNODES "ins_dnodes" #define TSDB_INS_TABLE_MNODES "ins_mnodes" From c9130fe71b302527b207460d2a1a3a5ef2224ff9 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Fri, 24 Feb 2023 15:17:37 +0800 Subject: [PATCH 34/34] fix: refine dockerfile for main (#20149) --- packaging/docker/DockerfileCloud | 30 ------------------------------ 1 file changed, 30 deletions(-) delete mode 100644 packaging/docker/DockerfileCloud diff --git a/packaging/docker/DockerfileCloud b/packaging/docker/DockerfileCloud deleted file mode 100644 index fa8fcabf34..0000000000 --- a/packaging/docker/DockerfileCloud +++ /dev/null @@ -1,30 +0,0 @@ -FROM ubuntu:18.04 - -WORKDIR /root - -ARG pkgFile -ARG dirName -ARG cpuType -RUN echo ${pkgFile} && echo ${dirName} - -RUN apt update -RUN apt install -y curl - -COPY ${pkgFile} /root/ -ENV TINI_VERSION v0.19.0 -ENV TAOS_DISABLE_ADAPTER 1 -ADD https://github.com/krallin/tini/releases/download/${TINI_VERSION}/tini-${cpuType} /tini -ENV DEBIAN_FRONTEND=noninteractive -WORKDIR /root/ -RUN tar -zxf ${pkgFile} && cd /root/${dirName}/ && /bin/bash install.sh -e no && cd /root && rm /root/${pkgFile} && rm -rf /root/${dirName} && apt-get update && apt-get install -y locales tzdata netcat && locale-gen en_US.UTF-8 && apt-get clean && rm -rf /var/lib/apt/lists/ && chmod +x /tini - -ENV LD_LIBRARY_PATH="$LD_LIBRARY_PATH:/usr/lib" \ - LC_CTYPE=en_US.UTF-8 \ - LANG=en_US.UTF-8 \ - LC_ALL=en_US.UTF-8 -COPY ./run.sh /usr/bin/ -COPY ./bin/* /usr/bin/ - -ENTRYPOINT ["/tini", "--", "/usr/bin/entrypoint.sh"] -CMD ["bash", "-c", "/usr/bin/run.sh"] -VOLUME [ "/var/lib/taos", "/var/log/taos" ]