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

merge main

This commit is contained in:
Xiaoyu Wang 2023-02-24 13:26:45 +08:00
parent 1ab144609e 51b3e50e87
commit 58fc265b80
43 changed files with 421 additions and 567 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/cmake.version
View File

@ -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)

2
cmake/taostools_CMakeLists.txt.in
View File

@ -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

4
docs/en/28-releases/01-tdengine.md
View File

@ -10,6 +10,10 @@ For TDengine 2.x installation packages by version, please visit [here](https://w
import Release from "/components/ReleaseV3";
## 3.0.2.6
<Release type="tdengine" version="3.0.2.6" />
## 3.0.2.5
<Release type="tdengine" version="3.0.2.5" />

4
docs/en/28-releases/02-tools.md
View File

@ -10,6 +10,10 @@ For other historical version installers, please visit [here](https://www.taosdat
import Release from "/components/ReleaseV3";
## 2.4.6
<Release type="tools" version="2.4.6" />
## 2.4.3
<Release type="tools" version="2.4.3" />

8
docs/zh/12-taos-sql/24-show.md
View File

@ -193,7 +193,7 @@ _block_dist: Total_Blocks=[5] Total_Size=[93.65 Kb] Average_size=[18.73 Kb] Comp
Total_Blocks: 表 d0 占用的 block 个数为 5 个
Total_Size. : 表 d0 所有 block 在文件中占用的大小为 93.65 KB
Total_Size: 表 d0 所有 block 在文件中占用的大小为 93.65 KB
Average_size: 平均每个 block 在文件中占用的空间大小为 18.73 KB
@ -204,15 +204,15 @@ Compression_Ratio: 数据压缩率 23.98%
_block_dist: Total_Rows=[20000] Inmem_Rows=[0] MinRows=[3616] MaxRows=[4096] Average_Rows=[4000]
Total_Rows: 统计表 d0 的所有行数 为20000 行
Total_Rows: 统计表 d0 的所有行数 为20000 行（该数值仅供参考，不是精确的行数。获得精确的行数需要使用 count 函数）
Inmem_Rows： 表示仍然还存放在内存中的行数，即没有落盘的行数，为 0行，表示没有
Inmem_Rows： 存储在写缓存中的数据行数（没有落盘），0 行表示内存缓存中没有数据
MinRows： BLOCK 中最小的行数，为 3616 行
MaxRows： BLOCK 中最大的行数，为 4096 行
Average_Rows： 每个 BLOCK 中的平均行数，为4000 行
Average_Rows： 每个 BLOCK 中的平均行数，此时为 4000 行
*************************** 3.row ***************************

4
docs/zh/28-releases/01-tdengine.md
View File

@ -10,6 +10,10 @@ TDengine 2.x 各版本安装包请访问[这里](https://www.taosdata.com/all-do
import Release from "/components/ReleaseV3";
## 3.0.2.6
<Release type="tdengine" version="3.0.2.6" />
## 3.0.2.5
<Release type="tdengine" version="3.0.2.5" />

4
docs/zh/28-releases/02-tools.md
View File

@ -10,6 +10,10 @@ taosTools 各版本安装包下载链接如下：
import Release from "/components/ReleaseV3";
## 2.4.6
<Release type="tools" version="2.4.6" />
## 2.4.3
<Release type="tools" version="2.4.3" />

24
examples/c/tmq.c
View File

@ -194,35 +194,43 @@ tmq_t* build_consumer() {
tmq_conf_t* conf = tmq_conf_new();
code = tmq_conf_set(conf, "enable.auto.commit", "true");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "group.id", "cgrpName");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "client.id", "user defined name");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "td.connect.user", "root");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "td.connect.pass", "taosdata");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "auto.offset.reset", "earliest");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
code = tmq_conf_set(conf, "experimental.snapshot.enable", "false");
if (TMQ_CONF_OK != code) {
goto _end;
tmq_conf_destroy(conf);
return NULL;
}
tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);

7
source/common/src/tdatablock.c
View File

@ -1067,6 +1067,7 @@ SHelper* createTupleIndex_rv(int32_t numOfRows, SArray* pOrderInfo, SSDataBlock*
offset += pInfo->pColData->info.bytes;
}
taosMemoryFree(buf);
return phelper;
}
@ -2404,7 +2405,11 @@ _end:
taosArrayDestroy(pVals);
if (terrno != 0) {
*ppReq = NULL;
if (pReq) tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
if (pReq) {
tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
taosMemoryFreeClear(pReq);
}
return TSDB_CODE_FAILED;
}
*ppReq = pReq;

103
source/common/src/tdataformat.c
View File

@ -1989,8 +1989,7 @@ static FORCE_INLINE int32_t tColDataUpdateValue20(SColData *pColData, uint8_t *p
return 0;
}
static FORCE_INLINE int32_t tColDataUpdateValue30(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
uint8_t bv = GET_BIT1(pColData->pBitMap, pColData->nVal - 1);
if (bv == 0) { // NONE == > VALUE
if (GET_BIT1(pColData->pBitMap, pColData->nVal - 1) == 0) { // NONE ==> VALUE
pColData->numOfNone--;
pColData->nVal--;
if (pColData->numOfNone) {
@ -2024,36 +2023,43 @@ static FORCE_INLINE int32_t tColDataUpdateValue32(SColData *pColData, uint8_t *p
return 0;
}
static FORCE_INLINE int32_t tColDataUpdateValue40(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
if (forward) {
pColData->numOfValue--;
if (forward) { // VALUE ==> VALUE
pColData->nVal--;
if (pColData->numOfValue) {
return tColDataAppendValue40(pColData, pData, nData);
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->flag = 0;
return tColDataAppendValue00(pColData, pData, nData);
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataPutValue(pColData, pData, nData);
}
return 0;
}
static FORCE_INLINE int32_t tColDataUpdateValue42(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
if (forward) {
if (forward) { // VALUE ==> NULL
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataAppendValue42(pColData, pData, nData);
} else {
pColData->flag = 0;
pColData->nData = 0;
return tColDataAppendValue02(pColData, pData, nData);
}
}
return 0;
}
static FORCE_INLINE int32_t tColDataUpdateValue50(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
uint8_t bv = GET_BIT1(pColData->pBitMap, pColData->nVal - 1);
if (bv == 0) { // NONE ==> VALUE
if (GET_BIT1(pColData->pBitMap, pColData->nVal - 1) == 0) { // NONE ==> VALUE
pColData->numOfNone--;
pColData->nVal--;
if (!IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData -= TYPE_BYTES[pColData->type];
}
if (pColData->numOfNone) {
return tColDataAppendValue50(pColData, pData, nData);
} else {
@ -2061,35 +2067,42 @@ static FORCE_INLINE int32_t tColDataUpdateValue50(SColData *pColData, uint8_t *p
return tColDataAppendValue40(pColData, pData, nData);
}
} else if (forward) { // VALUE ==> VALUE
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
return tColDataAppendValue50(pColData, pData, nData);
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->flag = HAS_NONE;
return tColDataAppendValue10(pColData, pData, nData);
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataPutValue(pColData, pData, nData);
}
return 0;
}
static FORCE_INLINE int32_t tColDataUpdateValue52(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
uint8_t bv = GET_BIT1(pColData->pBitMap, pColData->nVal - 1);
if (bv == 0) { // NONE ==> NULL
if (GET_BIT1(pColData->pBitMap, pColData->nVal - 1) == 0) { // NONE ==> NULL
pColData->numOfNone--;
pColData->nVal--;
if (!IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData -= TYPE_BYTES[pColData->type];
}
if (pColData->numOfNone) {
return tColDataAppendValue52(pColData, pData, nData);
} else {
pColData->flag &= ~HAS_NONE;
pColData->flag = HAS_VALUE;
return tColDataAppendValue42(pColData, pData, nData);
}
} else if (forward) { // VALUE ==> NULL
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataAppendValue52(pColData, pData, nData);
} else {
pColData->flag &= ~HAS_VALUE;
pColData->flag = HAS_NONE;
pColData->nData = 0;
return tColDataAppendValue12(pColData, pData, nData);
}
}
@ -2100,6 +2113,9 @@ static FORCE_INLINE int32_t tColDataUpdateValue60(SColData *pColData, uint8_t *p
if (GET_BIT1(pColData->pBitMap, pColData->nVal - 1) == 0) { // NULL ==> VALUE
pColData->numOfNull--;
pColData->nVal--;
if (!IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData -= TYPE_BYTES[pColData->type];
}
if (pColData->numOfNull) {
return tColDataAppendValue60(pColData, pData, nData);
} else {
@ -2107,14 +2123,13 @@ static FORCE_INLINE int32_t tColDataUpdateValue60(SColData *pColData, uint8_t *p
return tColDataAppendValue40(pColData, pData, nData);
}
} else { // VALUE ==> VALUE
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
return tColDataAppendValue60(pColData, pData, nData);
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->flag = HAS_NULL;
return tColDataAppendValue20(pColData, pData, nData);
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataPutValue(pColData, pData, nData);
}
}
return 0;
@ -2124,9 +2139,15 @@ static FORCE_INLINE int32_t tColDataUpdateValue62(SColData *pColData, uint8_t *p
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataAppendValue62(pColData, pData, nData);
} else {
pColData->flag = HAS_NULL;
pColData->nData = 0;
return tColDataAppendValue20(pColData, pData, nData);
}
}
@ -2139,38 +2160,44 @@ static FORCE_INLINE int32_t tColDataUpdateValue70(SColData *pColData, uint8_t *p
if (bv == 0) { // NONE ==> VALUE
pColData->numOfNone--;
pColData->nVal--;
if (!IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData -= TYPE_BYTES[pColData->type];
}
if (pColData->numOfNone) {
return tColDataAppendValue70(pColData, pData, nData);
} else {
for (int32_t iVal = 0; iVal < pColData->nVal; ++iVal) {
SET_BIT1(pColData->pBitMap, iVal, GET_BIT2(pColData->pBitMap, iVal) - 1);
}
pColData->flag &= ~HAS_NONE;
pColData->flag = (HAS_VALUE | HAS_NULL);
return tColDataAppendValue60(pColData, pData, nData);
}
} else if (bv == 1) { // NULL ==> VALUE
if (forward) {
pColData->numOfNull--;
pColData->nVal--;
if (!IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData -= TYPE_BYTES[pColData->type];
}
if (pColData->numOfNull) {
return tColDataAppendValue70(pColData, pData, nData);
} else {
for (int32_t iVal = 0; iVal < pColData->nVal; ++iVal) {
SET_BIT1(pColData->pBitMap, iVal, GET_BIT2(pColData->pBitMap, iVal) ? 1 : 0);
}
pColData->flag &= ~HAS_NULL;
pColData->flag = (HAS_VALUE | HAS_NONE);
return tColDataAppendValue50(pColData, pData, nData);
}
}
} else if (bv == 2) { // VALUE ==> VALUE
if (forward) {
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
return tColDataAppendValue70(pColData, pData, nData);
if (IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
return tColDataPutValue(pColData, pData, nData);
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataPutValue(pColData, pData, nData);
}
} else {
ASSERT(0);
@ -2179,29 +2206,37 @@ static FORCE_INLINE int32_t tColDataUpdateValue70(SColData *pColData, uint8_t *p
}
static int32_t tColDataUpdateValue72(SColData *pColData, uint8_t *pData, uint32_t nData, bool forward) {
uint8_t bv = GET_BIT2(pColData->pBitMap, pColData->nVal - 1);
ASSERT(bv < 3);
if (bv == 0) { // NONE ==> NULL
pColData->numOfNone--;
pColData->nVal--;
if (!IS_VAR_DATA_TYPE(pColData->type)) {
pColData->nData -= TYPE_BYTES[pColData->type];
}
if (pColData->numOfNone) {
return tColDataAppendValue72(pColData, pData, nData);
} else {
for (int32_t iVal = 0; iVal < pColData->nVal; ++iVal) {
SET_BIT1(pColData->pBitMap, iVal, GET_BIT2(pColData->pBitMap, iVal) - 1);
}
pColData->flag &= ~HAS_NONE;
pColData->flag = (HAS_VALUE | HAS_NULL);
return tColDataAppendValue62(pColData, pData, nData);
}
} else if (bv == 2 && forward) { // VALUE ==> NULL
pColData->numOfValue--;
pColData->nVal--;
if (pColData->numOfValue) {
if (IS_STR_DATA_TYPE(pColData->type)) {
pColData->nData = pColData->aOffset[pColData->nVal];
} else {
pColData->nData -= TYPE_BYTES[pColData->type];
}
return tColDataAppendValue72(pColData, pData, nData);
} else {
for (int32_t iVal = 0; iVal < pColData->nVal; ++iVal) {
SET_BIT1(pColData->pBitMap, iVal, GET_BIT2(pColData->pBitMap, iVal));
}
pColData->flag &= ~HAS_VALUE;
pColData->flag = (HAS_NULL | HAS_NONE);
pColData->nData = 0;
return tColDataAppendValue32(pColData, pData, nData);
}
}

12
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -93,6 +93,14 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
break;
}
/*
pDnode is null, TD-22618
at trans.c line 91
before this line, dmProcessRpcMsg callback is set
after this line, parent is set
so when dmProcessRpcMsg is called, pDonde is still null.
*/
if (pDnode != NULL){
if(pDnode->status != DND_STAT_RUNNING) {
if (pRpc->msgType == TDMT_DND_SERVER_STATUS) {
dmProcessServerStartupStatus(pDnode, pRpc);
@ -106,6 +114,10 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
goto _OVER;
}
}
} else {
terrno = TSDB_CODE_APP_IS_STARTING;
goto _OVER;
}
if (pRpc->pCont == NULL && (IsReq(pRpc) || pRpc->contLen != 0)) {
dGError("msg:%p, type:%s pCont is NULL", pRpc, TMSG_INFO(pRpc->msgType));

2
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -916,6 +916,8 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
if (!existing) {
taosArrayPush(pOldConsumer->currentTopics, &addedTopic);
taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString);
} else {
taosMemoryFree(addedTopic);
}
// set status

2
source/dnode/mnode/impl/src/mndDump.c
View File

@ -629,7 +629,7 @@ void mndDumpSdb() {
}
taosWriteFile(pFile, pCont, contLen);
taosWriteFile(pFile, "\n", 1);
taosFsyncFile(pFile);
UNUSED(taosFsyncFile(pFile));
taosCloseFile(&pFile);
tjsonDelete(json);
taosMemoryFree(pCont);

8
source/dnode/mnode/impl/src/mndStream.c
View File

@ -354,7 +354,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
int32_t dataIndex = 0;
for (int16_t i = 0; i < pObj->outputSchema.nCols; i++) {
SColLocation *pos = taosArrayGet(pCreate->fillNullCols, nullIndex);
if (i < pos->slotId) {
if (nullIndex >= numOfNULL || i < pos->slotId) {
pFullSchema[i].bytes = pObj->outputSchema.pSchema[dataIndex].bytes;
pFullSchema[i].colId = i + 1; // pObj->outputSchema.pSchema[dataIndex].colId;
pFullSchema[i].flags = pObj->outputSchema.pSchema[dataIndex].flags;
@ -723,8 +723,10 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
mInfo("trans:%d, used to create stream:%s", pTrans->id, createStreamReq.name);
mndTransSetDbName(pTrans, createStreamReq.sourceDB, streamObj.targetDb);
if (mndTrancCheckConflict(pMnode, pTrans) != 0) goto _OVER;
if (mndTrancCheckConflict(pMnode, pTrans) != 0) {
mndTransDrop(pTrans);
goto _OVER;
}
// create stb for stream
if (createStreamReq.createStb == STREAM_CREATE_STABLE_TRUE &&
mndCreateStbForStream(pMnode, pTrans, &streamObj, pReq->info.conn.user) < 0) {

2
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -706,7 +706,7 @@ int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sv
}
}
if (ASSERTS(sver > 0, __FILE__, __LINE__, "failed to get table schema version: %d", sver)) {
if (ASSERTS(sver > 0, "failed to get table schema version: %d", sver)) {
code = TSDB_CODE_NOT_FOUND;
goto _exit;
}

9
source/dnode/vnode/src/tq/tqSink.c
View File

@ -497,7 +497,7 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void*
taosArrayPush(tagArray, &tagVal);
}
}
pCreateTbReq->ctb.tagNum = size;
pCreateTbReq->ctb.tagNum = TMAX(size - UD_TAG_COLUMN_INDEX, 1);
STag* pTag = NULL;
tTagNew(tagArray, 1, false, &pTag);
@ -510,15 +510,12 @@ void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void*
pCreateTbReq->ctb.pTag = (uint8_t*)pTag;
// set table name
SColumnInfoData* pTbColInfo = taosArrayGet(pDataBlock->pDataBlock, UD_TABLE_NAME_COLUMN_INDEX);
if (colDataIsNull_s(pTbColInfo, rowId)) {
if (!pDataBlock->info.parTbName[0]) {
SColumnInfoData* pGpIdColInfo = taosArrayGet(pDataBlock->pDataBlock, UD_GROUPID_COLUMN_INDEX);
void* pGpIdData = colDataGetData(pGpIdColInfo, rowId);
pCreateTbReq->name = buildCtbNameByGroupId(stbFullName, *(uint64_t*)pGpIdData);
} else {
void* pTbData = colDataGetData(pTbColInfo, rowId);
pCreateTbReq->name = taosMemoryCalloc(1, varDataLen(pTbData) + 1);
memcpy(pCreateTbReq->name, varDataVal(pTbData), varDataLen(pTbData));
pCreateTbReq->name = strdup(pDataBlock->info.parTbName);
}
taosArrayPush(reqs.pArray, pCreateTbReq);
}

3
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -813,7 +813,8 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
if (!state->pBlockData) {
state->pBlockData = &state->blockData;
tBlockDataCreate(&state->blockData);
code = tBlockDataCreate(&state->blockData);
if (code) goto _err;
}
}
case SFSNEXTROW_BLOCKDATA:

2
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -433,9 +433,11 @@ _end:
tsdbUntakeReadSnap((STsdbReader*)pr, pr->pReadSnap, true);
taosThreadMutexUnlock(&pr->readerMutex);
if (pRes != NULL) {
for (int32_t j = 0; j < pr->numOfCols; ++j) {
taosMemoryFree(pRes[j]);
}
}
taosMemoryFree(pRes);
taosArrayDestroyEx(pLastCols, freeItem);

2
source/dnode/vnode/src/tsdb/tsdbDataIter.c
View File

@ -219,7 +219,7 @@ static int32_t tsdbDataFileDataIterNext(STsdbDataIter2* pIter, STsdbFilterInfo*
}
ASSERT(pIter->rowInfo.suid == pIter->dIter.bData.suid);
ASSERT(pIter->rowInfo.uid = pIter->dIter.bData.uid);
ASSERT(pIter->rowInfo.uid == pIter->dIter.bData.uid);
pIter->rowInfo.row = tsdbRowFromBlockData(&pIter->dIter.bData, pIter->dIter.iRow);
pIter->dIter.iRow++;
goto _exit;

2
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -151,7 +151,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) {
int64_t size = 0;
int64_t n;
TdFilePtr pFD;
char fname[TSDB_FILENAME_LEN];
char fname[TSDB_FILENAME_LEN] = {0};
char hdr[TSDB_FHDR_SIZE] = {0};
// truncate

4
source/dnode/vnode/src/tsdb/tsdbMemTable.c
View File

@ -473,7 +473,7 @@ static int32_t tbDataDoPut(SMemTable *pMemTable, STbData *pTbData, SMemSkipListN
int8_t forward) {
int32_t code = 0;
int8_t level;
SMemSkipListNode *pNode;
SMemSkipListNode *pNode = NULL;
SVBufPool *pPool = pMemTable->pTsdb->pVnode->inUse;
int64_t nSize;
@ -591,7 +591,9 @@ static int32_t tsdbInsertColDataToTable(SMemTable *pMemTable, STbData *pTbData,
pBlockData->aColData = vnodeBufPoolMalloc(pPool, sizeof(SColData) * pBlockData->nColData);
if (pBlockData->aColData == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
for (int32_t iColData = 0; iColData < pBlockData->nColData; ++iColData) {
code = tColDataCopy(&aColData[iColData + 1], &pBlockData->aColData[iColData], (xMallocFn)vnodeBufPoolMalloc, pPool);
if (code) goto _exit;

2
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -874,7 +874,7 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
pBlockNum->numOfBlocks += 1;
}
if (pScanInfo->pBlockList != NULL && taosArrayGetSize(pScanInfo->pBlockList) > 0) {
if ((pScanInfo->pBlockList != NULL )&& (taosArrayGetSize(pScanInfo->pBlockList) > 0)) {
numOfQTable += 1;
}
}

18
source/libs/executor/src/groupoperator.c
View File

@ -989,32 +989,42 @@ void appendCreateTableRow(SStreamState* pState, SExprSupp* pTableSup, SExprSupp*
SSDataBlock* pTmpBlock = blockCopyOneRow(pSrcBlock, rowId);
memset(pTmpBlock->info.parTbName, 0, TSDB_TABLE_NAME_LEN);
pTmpBlock->info.id.groupId = groupId;
char* tbName = pSrcBlock->info.parTbName;
if (pTableSup->numOfExprs > 0) {
projectApplyFunctions(pTableSup->pExprInfo, pDestBlock, pTmpBlock, pTableSup->pCtx, pTableSup->numOfExprs, NULL);
SColumnInfoData* pTbCol = taosArrayGet(pDestBlock->pDataBlock, UD_TABLE_NAME_COLUMN_INDEX);
void* pData = colDataGetVarData(pTbCol, pDestBlock->info.rows - 1);
char* tbName = pSrcBlock->info.parTbName;
memset(tbName, 0, TSDB_TABLE_NAME_LEN);
int32_t len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN - 1);
int32_t len = 0;
if (colDataIsNull_s(pTbCol, pDestBlock->info.rows - 1)) {
len = TMIN(sizeof(TSDB_DATA_NULL_STR), TSDB_TABLE_NAME_LEN - 1);
memcpy(tbName, TSDB_DATA_NULL_STR, len);
} else {
void* pData = colDataGetData(pTbCol, pDestBlock->info.rows - 1);
len = TMIN(varDataLen(pData), TSDB_TABLE_NAME_LEN - 1);
memcpy(tbName, varDataVal(pData), len);
}
streamStatePutParName(pState, groupId, tbName);
memcpy(pTmpBlock->info.parTbName, tbName, len);
pDestBlock->info.rows--;
} else {
void* pTbNameCol = taosArrayGet(pDestBlock->pDataBlock, UD_TABLE_NAME_COLUMN_INDEX);
colDataSetNULL(pTbNameCol, pDestBlock->info.rows);
pSrcBlock->info.parTbName[0] = 0;
tbName[0] = 0;
}
if (pTagSup->numOfExprs > 0) {
projectApplyFunctions(pTagSup->pExprInfo, pDestBlock, pTmpBlock, pTagSup->pCtx, pTagSup->numOfExprs, NULL);
pDestBlock->info.rows--;
} else {
memcpy(pDestBlock->info.parTbName, pTmpBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
}
void* pGpIdCol = taosArrayGet(pDestBlock->pDataBlock, UD_GROUPID_COLUMN_INDEX);
colDataAppend(pGpIdCol, pDestBlock->info.rows, (const char*)&groupId, false);
pDestBlock->info.rows++;
blockDataDestroy(pTmpBlock);
} else {
memcpy(pSrcBlock->info.parTbName, pValue, TSDB_TABLE_NAME_LEN);
}
streamStateReleaseBuf(pState, NULL, pValue);
}

1
source/libs/executor/src/projectoperator.c
View File

@ -275,7 +275,6 @@ SSDataBlock* doProjectOperation(SOperatorInfo* pOperator) {
// for stream interval
if (pBlock->info.type == STREAM_RETRIEVE || pBlock->info.type == STREAM_DELETE_RESULT ||
pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
// printDataBlock1(pBlock, "project1");
return pBlock;
}

9
source/libs/executor/src/sortoperator.c
View File

@ -47,7 +47,8 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
SDataBlockDescNode* pDescNode = pSortNode->node.pOutputDataBlockDesc;
int32_t numOfCols = 0;
SExprInfo* pExprInfo = createExprInfo(pSortNode->pExprs, NULL, &numOfCols);
pOperator->exprSupp.pExprInfo = createExprInfo(pSortNode->pExprs, NULL, &numOfCols);
pOperator->exprSupp.numOfExprs = numOfCols;
int32_t numOfOutputCols = 0;
int32_t code =
@ -56,7 +57,8 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
goto _error;
}
pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pOperator->exprSupp.rowEntryInfoOffset);
pOperator->exprSupp.pCtx =
createSqlFunctionCtx(pOperator->exprSupp.pExprInfo, numOfCols, &pOperator->exprSupp.rowEntryInfoOffset);
initResultSizeInfo(&pOperator->resultInfo, 1024);
code = filterInitFromNode((SNode*)pSortNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
if (code != TSDB_CODE_SUCCESS) {
@ -68,8 +70,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
initLimitInfo(pSortNode->node.pLimit, pSortNode->node.pSlimit, &pInfo->limitInfo);
setOperatorInfo(pOperator, "SortOperator", QUERY_NODE_PHYSICAL_PLAN_SORT, true, OP_NOT_OPENED, pInfo, pTaskInfo);
pOperator->exprSupp.pExprInfo = pExprInfo;
pOperator->exprSupp.numOfExprs = numOfCols;
// lazy evaluation for the following parameter since the input datablock is not known till now.
// pInfo->bufPageSize = rowSize < 1024 ? 1024 * 2 : rowSize * 2;

3
source/libs/executor/src/sysscanoperator.c
View File

@ -1607,8 +1607,7 @@ static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScan
if (pInfo->tbnameSlotId != -1) {
SColumnInfoData* pColumnInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, pInfo->tbnameSlotId);
char varTbName[TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE] = {0};
memcpy(varDataVal(varTbName), name, strlen(name));
varDataSetLen(varTbName, strlen(name));
STR_TO_VARSTR(varTbName, name);
colDataSetNItems(pColumnInfoData, 0, varTbName, pBlock->info.rows);
}

1
source/libs/executor/src/tsort.c
View File

@ -212,6 +212,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
int32_t pageId = -1;
void* pPage = getNewBufPage(pHandle->pBuf, &pageId);
if (pPage == NULL) {
taosArrayDestroy(pPageIdList);
blockDataDestroy(p);
taosArrayDestroy(pPageIdList);
return terrno;

7
source/libs/function/src/builtinsimpl.c
View File

@ -1686,11 +1686,7 @@ int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
double v = 0;
tMemBucket* pMemBucket = ppInfo->pMemBucket;
if (pMemBucket == NULL || pMemBucket->total == 0) { // check for null
code = TSDB_CODE_FAILED;
goto _fin_error;
}
if (pMemBucket != NULL && pMemBucket->total > 0) { // check for null
if (pCtx->numOfParams > 2) {
char buf[512] = {0};
size_t len = 1;
@ -1734,6 +1730,7 @@ int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
tMemBucketDestroy(pMemBucket);
return functionFinalize(pCtx, pBlock);
}
}
_fin_error:

2
source/libs/sync/inc/syncPipeline.h
View File

@ -68,7 +68,7 @@ void syncNodeLogReplMgrDestroy(SSyncNode* pNode);
// access
static FORCE_INLINE int64_t syncLogGetRetryBackoffTimeMs(SSyncLogReplMgr* pMgr) {
return (1 << pMgr->retryBackoff) * SYNC_LOG_REPL_RETRY_WAIT_MS;
return ((int64_t)1 << pMgr->retryBackoff) * SYNC_LOG_REPL_RETRY_WAIT_MS;
}
static FORCE_INLINE int32_t syncLogGetNextRetryBackoff(SSyncLogReplMgr* pMgr) {

33
source/libs/sync/inc/syncRaftEntry.h
View File

@ -49,39 +49,6 @@ static FORCE_INLINE bool syncLogIsReplicationBarrier(SSyncRaftEntry* pEntry) {
return pEntry->originalRpcType == TDMT_SYNC_NOOP;
}
typedef struct SRaftEntryHashCache {
SHashObj* pEntryHash;
int32_t maxCount;
int32_t currentCount;
TdThreadMutex mutex;
SSyncNode* pSyncNode;
} SRaftEntryHashCache;
SRaftEntryHashCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount);
void raftCacheDestroy(SRaftEntryHashCache* pCache);
int32_t raftCachePutEntry(struct SRaftEntryHashCache* pCache, SSyncRaftEntry* pEntry);
int32_t raftCacheGetEntry(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheGetEntryP(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheDelEntry(struct SRaftEntryHashCache* pCache, SyncIndex index);
int32_t raftCacheGetAndDel(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftCacheClear(struct SRaftEntryHashCache* pCache);
typedef struct SRaftEntryCache {
SSkipList* pSkipList;
int32_t maxCount;
int32_t currentCount;
int32_t refMgr;
TdThreadMutex mutex;
SSyncNode* pSyncNode;
} SRaftEntryCache;
SRaftEntryCache* raftEntryCacheCreate(SSyncNode* pSyncNode, int32_t maxCount);
void raftEntryCacheDestroy(SRaftEntryCache* pCache);
int32_t raftEntryCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry);
int32_t raftEntryCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry);
int32_t raftEntryCacheClear(struct SRaftEntryCache* pCache, int32_t count);
#ifdef __cplusplus
}
#endif

8
source/libs/sync/src/syncAppendEntries.c
View File

@ -104,6 +104,8 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncAppendEntries* pMsg = pRpcMsg->pCont;
SRpcMsg rpcRsp = {0};
bool accepted = false;
SSyncRaftEntry* pEntry = NULL;
// if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) {
syncLogRecvAppendEntries(ths, pMsg, "not in my config");
@ -137,14 +139,13 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
syncNodeStepDown(ths, pMsg->term);
syncNodeResetElectTimer(ths);
if (pMsg->dataLen < (int32_t)sizeof(SSyncRaftEntry)) {
if (pMsg->dataLen < sizeof(SSyncRaftEntry)) {
sError("vgId:%d, incomplete append entries received. prev index:%" PRId64 ", term:%" PRId64 ", datalen:%d",
ths->vgId, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->dataLen);
goto _IGNORE;
}
SSyncRaftEntry* pEntry = syncBuildRaftEntryFromAppendEntries(pMsg);
pEntry = syncBuildRaftEntryFromAppendEntries(pMsg);
if (pEntry == NULL) {
sError("vgId:%d, failed to get raft entry from append entries since %s", ths->vgId, terrstr());
goto _IGNORE;
@ -191,5 +192,6 @@ _out:
_IGNORE:
rpcFreeCont(rpcRsp.pCont);
syncEntryDestroy(pEntry);
return 0;
}

2
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -40,7 +40,7 @@
//
int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncAppendEntriesReply* pMsg = pRpcMsg->pCont;
SyncAppendEntriesReply* pMsg = (SyncAppendEntriesReply*)pRpcMsg->pCont;
int32_t ret = 0;
// if already drop replica, do not process

24
source/libs/sync/src/syncMain.c
View File

@ -1126,7 +1126,10 @@ int32_t syncNodeStartStandBy(SSyncNode* pSyncNode) {
}
void syncNodePreClose(SSyncNode* pSyncNode) {
if (pSyncNode != NULL && pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpApplyQueueItems != NULL) {
ASSERT(pSyncNode != NULL);
ASSERT(pSyncNode->pFsm != NULL);
ASSERT(pSyncNode->pFsm->FpApplyQueueItems != NULL);
while (1) {
int32_t aqItems = pSyncNode->pFsm->FpApplyQueueItems(pSyncNode->pFsm);
sTrace("vgId:%d, pre close, %d items in apply queue", pSyncNode->vgId, aqItems);
@ -1135,20 +1138,6 @@ void syncNodePreClose(SSyncNode* pSyncNode) {
}
taosMsleep(20);
}
}
#if 0
if (pSyncNode->pNewNodeReceiver != NULL) {
if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
snapshotReceiverStop(pSyncNode->pNewNodeReceiver);
}
sDebug("vgId:%d, snapshot receiver destroy while preclose sync node, data:%p", pSyncNode->vgId,
pSyncNode->pNewNodeReceiver);
snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
pSyncNode->pNewNodeReceiver = NULL;
}
#endif
// stop elect timer
syncNodeStopElectTimer(pSyncNode);
@ -1461,7 +1450,7 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
}
// log begin config change
sNInfo(pSyncNode, "begin do config change, from %d to %d", pSyncNode->vgId, oldConfig.replicaNum,
sNInfo(pSyncNode, "begin do config change, from %d to %d, replicas:%d", pSyncNode->vgId, oldConfig.replicaNum,
pNewConfig->replicaNum);
if (IamInNew) {
@ -1742,8 +1731,7 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
#endif
// close receiver
if (pSyncNode != NULL && pSyncNode->pNewNodeReceiver != NULL &&
snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
snapshotReceiverStop(pSyncNode->pNewNodeReceiver);
}

6
source/libs/sync/src/syncPipeline.c
View File

@ -910,7 +910,7 @@ int32_t syncLogReplMgrProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNo
int64_t firstSentMs = pMgr->states[pMgr->startIndex % pMgr->size].timeMs;
int64_t lastSentMs = pMgr->states[(pMgr->endIndex - 1) % pMgr->size].timeMs;
int64_t timeDiffMs = lastSentMs - firstSentMs;
if (timeDiffMs > 0 && timeDiffMs < (SYNC_LOG_REPL_RETRY_WAIT_MS << (pMgr->retryBackoff - 1))) {
if (timeDiffMs > 0 && timeDiffMs < ((int64_t)SYNC_LOG_REPL_RETRY_WAIT_MS << (pMgr->retryBackoff - 1))) {
pMgr->retryBackoff -= 1;
}
}
@ -937,10 +937,6 @@ SSyncLogReplMgr* syncLogReplMgrCreate() {
ASSERT(pMgr->size == TSDB_SYNC_LOG_BUFFER_SIZE);
return pMgr;
_err:
taosMemoryFree(pMgr);
return NULL;
}
void syncLogReplMgrDestroy(SSyncLogReplMgr* pMgr) {

2
source/libs/sync/src/syncRaftCfg.c
View File

@ -224,7 +224,7 @@ _OVER:
int32_t syncAddCfgIndex(SSyncNode *pNode, SyncIndex cfgIndex) {
SRaftCfg *pCfg = &pNode->raftCfg;
if (pCfg->configIndexCount <= MAX_CONFIG_INDEX_COUNT) {
if (pCfg->configIndexCount < MAX_CONFIG_INDEX_COUNT) {
return -1;
}

341
source/libs/sync/src/syncRaftEntry.c
View File

@ -102,344 +102,3 @@ void syncEntry2OriginalRpc(const SSyncRaftEntry* pEntry, SRpcMsg* pRpcMsg) {
pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
memcpy(pRpcMsg->pCont, pEntry->data, pRpcMsg->contLen);
}
SRaftEntryHashCache* raftCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
SRaftEntryHashCache* pCache = taosMemoryMalloc(sizeof(SRaftEntryHashCache));
if (pCache == NULL) {
sError("vgId:%d, raft cache create error", pSyncNode->vgId);
return NULL;
}
pCache->pEntryHash =
taosHashInit(sizeof(SyncIndex), taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (pCache->pEntryHash == NULL) {
sError("vgId:%d, raft cache create hash error", pSyncNode->vgId);
return NULL;
}
taosThreadMutexInit(&(pCache->mutex), NULL);
pCache->maxCount = maxCount;
pCache->currentCount = 0;
pCache->pSyncNode = pSyncNode;
return pCache;
}
void raftCacheDestroy(SRaftEntryHashCache* pCache) {
if (pCache != NULL) {
taosThreadMutexLock(&pCache->mutex);
taosHashCleanup(pCache->pEntryHash);
taosThreadMutexUnlock(&pCache->mutex);
taosThreadMutexDestroy(&(pCache->mutex));
taosMemoryFree(pCache);
}
}
// success, return 1
// max count, return 0
// error, return -1
int32_t raftCachePutEntry(struct SRaftEntryHashCache* pCache, SSyncRaftEntry* pEntry) {
taosThreadMutexLock(&pCache->mutex);
if (pCache->currentCount >= pCache->maxCount) {
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosHashPut(pCache->pEntryHash, &(pEntry->index), sizeof(pEntry->index), pEntry, pEntry->bytes);
++(pCache->currentCount);
sNTrace(pCache->pSyncNode, "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
taosThreadMutexUnlock(&pCache->mutex);
return 1;
}
// success, return 0
// error, return -1
// not exist, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
int32_t raftCacheGetEntry(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
if (ppEntry == NULL) {
return -1;
}
*ppEntry = NULL;
taosThreadMutexLock(&pCache->mutex);
void* pTmp = taosHashGet(pCache->pEntryHash, &index, sizeof(index));
if (pTmp != NULL) {
SSyncRaftEntry* pEntry = pTmp;
*ppEntry = taosMemoryMalloc(pEntry->bytes);
memcpy(*ppEntry, pTmp, pEntry->bytes);
sNTrace(pCache->pSyncNode, "raft cache get, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosThreadMutexUnlock(&pCache->mutex);
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
}
// success, return 0
// error, return -1
// not exist, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
int32_t raftCacheGetEntryP(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
if (ppEntry == NULL) {
return -1;
}
*ppEntry = NULL;
taosThreadMutexLock(&pCache->mutex);
void* pTmp = taosHashGet(pCache->pEntryHash, &index, sizeof(index));
if (pTmp != NULL) {
SSyncRaftEntry* pEntry = pTmp;
*ppEntry = pEntry;
sNTrace(pCache->pSyncNode, "raft cache get, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosThreadMutexUnlock(&pCache->mutex);
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
}
int32_t raftCacheDelEntry(struct SRaftEntryHashCache* pCache, SyncIndex index) {
taosThreadMutexLock(&pCache->mutex);
taosHashRemove(pCache->pEntryHash, &index, sizeof(index));
--(pCache->currentCount);
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
int32_t raftCacheGetAndDel(struct SRaftEntryHashCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
if (ppEntry == NULL) {
return -1;
}
*ppEntry = NULL;
taosThreadMutexLock(&pCache->mutex);
void* pTmp = taosHashGet(pCache->pEntryHash, &index, sizeof(index));
if (pTmp != NULL) {
SSyncRaftEntry* pEntry = pTmp;
*ppEntry = taosMemoryMalloc(pEntry->bytes);
memcpy(*ppEntry, pTmp, pEntry->bytes);
sNTrace(pCache->pSyncNode, "raft cache get-and-del, type:%s,%d, type2:%s,%d, index:%" PRId64,
TMSG_INFO((*ppEntry)->msgType), (*ppEntry)->msgType, TMSG_INFO((*ppEntry)->originalRpcType),
(*ppEntry)->originalRpcType, (*ppEntry)->index);
taosHashRemove(pCache->pEntryHash, &index, sizeof(index));
--(pCache->currentCount);
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
taosThreadMutexUnlock(&pCache->mutex);
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
}
int32_t raftCacheClear(struct SRaftEntryHashCache* pCache) {
taosThreadMutexLock(&pCache->mutex);
taosHashClear(pCache->pEntryHash);
pCache->currentCount = 0;
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
static char* keyFn(const void* pData) {
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)pData;
return (char*)(&(pEntry->index));
}
static int cmpFn(const void* p1, const void* p2) { return memcmp(p1, p2, sizeof(SyncIndex)); }
static void freeRaftEntry(void* param) {
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)param;
syncEntryDestroy(pEntry);
}
SRaftEntryCache* raftEntryCacheCreate(SSyncNode* pSyncNode, int32_t maxCount) {
SRaftEntryCache* pCache = taosMemoryMalloc(sizeof(SRaftEntryCache));
if (pCache == NULL) {
sError("vgId:%d, raft cache create error", pSyncNode->vgId);
return NULL;
}
pCache->pSkipList =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, sizeof(SyncIndex), cmpFn, SL_ALLOW_DUP_KEY, keyFn);
if (pCache->pSkipList == NULL) {
sError("vgId:%d, raft cache create hash error", pSyncNode->vgId);
return NULL;
}
taosThreadMutexInit(&(pCache->mutex), NULL);
pCache->refMgr = taosOpenRef(10, freeRaftEntry);
pCache->maxCount = maxCount;
pCache->currentCount = 0;
pCache->pSyncNode = pSyncNode;
return pCache;
}
void raftEntryCacheDestroy(SRaftEntryCache* pCache) {
if (pCache != NULL) {
taosThreadMutexLock(&pCache->mutex);
tSkipListDestroy(pCache->pSkipList);
if (pCache->refMgr != -1) {
taosCloseRef(pCache->refMgr);
pCache->refMgr = -1;
}
taosThreadMutexUnlock(&pCache->mutex);
taosThreadMutexDestroy(&(pCache->mutex));
taosMemoryFree(pCache);
}
}
// success, return 1
// max count, return 0
// error, return -1
int32_t raftEntryCachePutEntry(struct SRaftEntryCache* pCache, SSyncRaftEntry* pEntry) {
taosThreadMutexLock(&pCache->mutex);
if (pCache->currentCount >= pCache->maxCount) {
taosThreadMutexUnlock(&pCache->mutex);
return 0;
}
SSkipListNode* pSkipListNode = tSkipListPut(pCache->pSkipList, pEntry);
ASSERT(pSkipListNode != NULL);
++(pCache->currentCount);
pEntry->rid = taosAddRef(pCache->refMgr, pEntry);
ASSERT(pEntry->rid >= 0);
sNTrace(pCache->pSyncNode, "raft cache add, type:%s,%d, type2:%s,%d, index:%" PRId64 ", bytes:%d",
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType,
pEntry->index, pEntry->bytes);
taosThreadMutexUnlock(&pCache->mutex);
return 1;
}
// find one, return 1
// not found, return 0
// error, return -1
int32_t raftEntryCacheGetEntry(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
ASSERT(ppEntry != NULL);
SSyncRaftEntry* pEntry = NULL;
int32_t code = raftEntryCacheGetEntryP(pCache, index, &pEntry);
if (code == 1) {
int32_t bytes = (int32_t)pEntry->bytes;
*ppEntry = taosMemoryMalloc((int64_t)bytes);
memcpy(*ppEntry, pEntry, pEntry->bytes);
(*ppEntry)->rid = -1;
} else {
*ppEntry = NULL;
}
return code;
}
// find one, return 1
// not found, return 0
// error, return -1
int32_t raftEntryCacheGetEntryP(struct SRaftEntryCache* pCache, SyncIndex index, SSyncRaftEntry** ppEntry) {
taosThreadMutexLock(&pCache->mutex);
SyncIndex index2 = index;
int32_t code = 0;
SArray* entryPArray = tSkipListGet(pCache->pSkipList, (char*)(&index2));
int32_t arraySize = taosArrayGetSize(entryPArray);
if (arraySize == 1) {
SSkipListNode** ppNode = (SSkipListNode**)taosArrayGet(entryPArray, 0);
ASSERT(*ppNode != NULL);
*ppEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(*ppNode);
taosAcquireRef(pCache->refMgr, (*ppEntry)->rid);
code = 1;
} else if (arraySize == 0) {
code = 0;
} else {
ASSERT(0);
code = -1;
}
taosArrayDestroy(entryPArray);
taosThreadMutexUnlock(&pCache->mutex);
return code;
}
// count = -1, clear all
// count >= 0, clear count
// return -1, error
// return delete count
int32_t raftEntryCacheClear(struct SRaftEntryCache* pCache, int32_t count) {
taosThreadMutexLock(&pCache->mutex);
int32_t returnCnt = 0;
if (count == -1) {
// clear all
SSkipListIterator* pIter = tSkipListCreateIter(pCache->pSkipList);
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
syncEntryDestroy(pEntry);
++returnCnt;
}
tSkipListDestroyIter(pIter);
tSkipListDestroy(pCache->pSkipList);
pCache->pSkipList =
tSkipListCreate(MAX_SKIP_LIST_LEVEL, TSDB_DATA_TYPE_BINARY, sizeof(SyncIndex), cmpFn, SL_ALLOW_DUP_KEY, keyFn);
ASSERT(pCache->pSkipList != NULL);
} else {
// clear count
int i = 0;
SSkipListIterator* pIter = tSkipListCreateIter(pCache->pSkipList);
SArray* delNodeArray = taosArrayInit(0, sizeof(SSkipListNode*));
// free entry
while (tSkipListIterNext(pIter)) {
SSkipListNode* pNode = tSkipListIterGet(pIter);
ASSERT(pNode != NULL);
if (i++ >= count) {
break;
}
// sDebug("push pNode:%p", pNode);
taosArrayPush(delNodeArray, &pNode);
++returnCnt;
SSyncRaftEntry* pEntry = (SSyncRaftEntry*)SL_GET_NODE_DATA(pNode);
// syncEntryDestroy(pEntry);
taosRemoveRef(pCache->refMgr, pEntry->rid);
}
tSkipListDestroyIter(pIter);
// delete skiplist node
int32_t arraySize = taosArrayGetSize(delNodeArray);
for (int32_t i = 0; i < arraySize; ++i) {
SSkipListNode** ppNode = taosArrayGet(delNodeArray, i);
// sDebug("get pNode:%p", *ppNode);
tSkipListRemoveNode(pCache->pSkipList, *ppNode);
}
taosArrayDestroy(delNodeArray);
}
pCache->currentCount -= returnCnt;
taosThreadMutexUnlock(&pCache->mutex);
return returnCnt;
}

21
source/libs/sync/src/syncSnapshot.c
View File

@ -168,17 +168,19 @@ static int32_t snapshotSend(SSyncSnapshotSender *pSender) {
if (pSender->blockLen > 0) {
// has read data
sSDebug(pSender, "snapshot sender continue to read, blockLen:%d seq:%d", pSender->blockLen, pSender->seq);
sSDebug(pSender, "vgId:%d, snapshot sender continue to read, blockLen:%d seq:%d", pSender->pSyncNode->vgId,
pSender->blockLen, pSender->seq);
} else {
// read finish, update seq to end
pSender->seq = SYNC_SNAPSHOT_SEQ_END;
sSInfo(pSender, "snapshot sender read to the end, blockLen:%d seq:%d", pSender->blockLen, pSender->seq);
sSInfo(pSender, "vgId:%d, snapshot sender read to the end, blockLen:%d seq:%d", pSender->pSyncNode->vgId,
pSender->blockLen, pSender->seq);
}
// build msg
SRpcMsg rpcMsg = {0};
if (syncBuildSnapshotSend(&rpcMsg, pSender->blockLen, pSender->pSyncNode->vgId) != 0) {
sSError(pSender, "snapshot sender build msg failed since %s", pSender->pSyncNode->vgId, terrstr());
sSError(pSender, "vgId:%d, snapshot sender build msg failed since %s", pSender->pSyncNode->vgId, terrstr());
return -1;
}
@ -340,11 +342,13 @@ void snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver) {
taosMemoryFree(pReceiver);
}
bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) { return pReceiver->start; }
bool snapshotReceiverIsStart(SSyncSnapshotReceiver *pReceiver) {
return (pReceiver != NULL ? pReceiver->start : false);
}
static int32_t snapshotReceiverStartWriter(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pBeginMsg) {
if (pReceiver->pWriter != NULL) {
sRError(pReceiver, "vgId:%d, snapshot receiver writer is not null");
sRError(pReceiver, "vgId:%d, snapshot receiver writer is not null", pReceiver->pSyncNode->vgId);
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
}
@ -851,8 +855,8 @@ static int32_t syncNodeOnSnapshotPrepRsp(SSyncNode *pSyncNode, SSyncSnapshotSend
pMsg->snapBeginIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm);
if (pMsg->snapBeginIndex > snapshot.lastApplyIndex) {
sSError(pSender, "prepare snapshot failed since beginIndex:%d larger than applyIndex:%d", pMsg->snapBeginIndex,
snapshot.lastApplyIndex);
sSError(pSender, "prepare snapshot failed since beginIndex:%" PRId64 " larger than applyIndex:%" PRId64,
pMsg->snapBeginIndex, snapshot.lastApplyIndex);
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
return -1;
}
@ -966,7 +970,8 @@ int32_t syncNodeOnSnapshotRsp(SSyncNode *pSyncNode, const SRpcMsg *pRpcMsg) {
if (pSender->pReader == NULL || pSender->finish) {
syncLogRecvSyncSnapshotRsp(pSyncNode, pMsg, "snapshot sender invalid");
sSError(pSender, "snapshot sender invalid, pReader:%p finish:%d", pMsg->code, pSender->pReader, pSender->finish);
sSError(pSender, "snapshot sender invalid error:%s 0x%x, pReader:%p finish:%d", tstrerror(pMsg->code), pMsg->code,
pSender->pReader, pSender->finish);
terrno = pMsg->code;
goto _ERROR;
}

2
source/libs/wal/src/walMeta.c
View File

@ -913,7 +913,7 @@ int walLoadMeta(SWal* pWal) {
int64_t fileSize = 0;
taosStatFile(fnameStr, &fileSize, NULL);
if (fileSize == 0) {
taosRemoveFile(fnameStr);
(void)taosRemoveFile(fnameStr);
wDebug("vgId:%d, wal find empty meta ver %d", pWal->cfg.vgId, metaVer);
return -1;
}

2
source/libs/wal/src/walWrite.c
View File

@ -63,7 +63,7 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
wInfo("vgId:%d, restore from snapshot, remove file %s", pWal->cfg.vgId, fnameStr);
}
}
walRemoveMeta(pWal);
(void)walRemoveMeta(pWal);
pWal->writeCur = -1;
pWal->totSize = 0;

16
source/util/src/tpagedbuf.c
View File

@ -391,7 +391,7 @@ _error:
return TSDB_CODE_OUT_OF_MEMORY;
}
static char* doExtractPage(SDiskbasedBuf* pBuf) {
static char* doExtractPage(SDiskbasedBuf* pBuf, bool* newPage) {
char* availablePage = NULL;
if (NO_IN_MEM_AVAILABLE_PAGES(pBuf)) {
availablePage = evictBufPage(pBuf);
@ -405,6 +405,7 @@ static char* doExtractPage(SDiskbasedBuf* pBuf) {
if (availablePage == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
}
*newPage = true;
}
return availablePage;
@ -413,7 +414,8 @@ static char* doExtractPage(SDiskbasedBuf* pBuf) {
void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
pBuf->statis.getPages += 1;
char* availablePage = doExtractPage(pBuf);
bool newPage = false;
char* availablePage = doExtractPage(pBuf, &newPage);
if (availablePage == NULL) {
return NULL;
}
@ -432,6 +434,9 @@ void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
// register page id info
pi = registerNewPageInfo(pBuf, *pageId);
if (pi == NULL) {
if (newPage) {
taosMemoryFree(availablePage);
}
return NULL;
}
@ -492,7 +497,8 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
ASSERT((!BUF_PAGE_IN_MEM(*pi)) && (*pi)->pn == NULL &&
(((*pi)->length >= 0 && (*pi)->offset >= 0) || ((*pi)->length == -1 && (*pi)->offset == -1)));
(*pi)->pData = doExtractPage(pBuf);
bool newPage = false;
(*pi)->pData = doExtractPage(pBuf, &newPage);
// failed to evict buffer page, return with error code.
if ((*pi)->pData == NULL) {
@ -509,6 +515,10 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
if (HAS_DATA_IN_DISK(*pi)) {
int32_t code = loadPageFromDisk(pBuf, *pi);
if (code != 0) {
if (newPage) {
taosMemoryFree((*pi)->pData);
}
terrno = code;
return NULL;
}

155
tests/script/tsim/stream/checkStreamSTable.sim
View File

@ -301,7 +301,7 @@ print $data00, $data01, $data02, $data03
print $data10, $data11, $data12, $data13
print $data20, $data21, $data22, $data23
loop2:
loop3:
sleep 300
@ -317,47 +317,182 @@ if $rows != 2 then
print $data00, $data01, $data02, $data03
print $data10, $data11, $data12, $data13
print $data20, $data21, $data22, $data23
goto loop2
goto loop3
endi
if $data01 != 10 then
print =====data01=$data01
goto loop2
goto loop3
endi
if $data02 != 20 then
print =====data02=$data02
goto loop2
goto loop3
endi
if $data03 != 1 then
print =====data03=$data03
goto loop2
goto loop3
endi
if $data04 != NULL then
print =====data04=$data04
goto loop2
goto loop3
endi
if $data11 != 40 then
print =====data11=$data11
goto loop2
goto loop3
endi
if $data12 != 50 then
print =====data12=$data12
goto loop2
goto loop3
endi
if $data13 != 1 then
print =====data13=$data13
goto loop2
goto loop3
endi
if $data14 != NULL then
print =====data14=$data14
goto loop2
goto loop3
endi
print ===== step7
sql create database result5 vgroups 1;
sql create database test5 vgroups 4;
sql use test5;
sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int);
sql create table t1 using st tags(1,2,3);
sql create table t2 using st tags(4,5,6);
sql create stable result5.streamt5(ts timestamp,a int,b int,c int, d int) tags(tg1 int,tg2 int,tg3 int);
sql create stream streams5 trigger at_once into result5.streamt5(ts,c,a,b) tags(tg2, tg3, tg1) subtable( concat("tbl-", cast(tg3 as varchar(10)) ) ) as select _wstart, count(*) c1, max(a),min(b) c2 from st partition by ta+1 as tg1, cast(tb as bigint) as tg2, a as tg3 session(ts, 10s);
sql insert into t1 values(1648791213000,NULL,NULL,NULL);
$loop_count = 0
print select _wstart, count(*) c1, max(a),min(b) c2 from st partition by ta+1 as tg1, cast(tb as bigint) as tg2, a as tg3 session(ts, 10s);
sql select _wstart, count(*) c1, max(a),min(b) c2 from st partition by ta+1 as tg1, cast(tb as bigint) as tg2, a as tg3 session(ts, 10s);
print $data00, $data01, $data02, $data03
print $data10, $data11, $data12, $data13
print $data20, $data21, $data22, $data23
loop4:
sleep 300
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
print sql select * from result5.streamt5 order by tg1;
sql select * from result5.streamt5 order by tg1;
print $data00, $data01, $data02, $data03 $data04 $data05 $data06 $data07
print $data10, $data11, $data12, $data13
print $data20, $data21, $data22, $data23
if $rows != 1 then
print =====rows=$rows
goto loop4
endi
if $data01 != NULL then
print =====data01=$data01
goto loop4
endi
if $data02 != NULL then
print =====data02=$data02
goto loop4
endi
if $data03 != 1 then
print =====data03=$data03
goto loop4
endi
if $data04 != NULL then
print =====data04=$data04
goto loop4
endi
if $data05 != 2 then
print =====data05=$data05
goto loop4
endi
if $data06 != 2 then
print =====data06=$data06
goto loop4
endi
if $data07 != NULL then
print =====data07=$data07
goto loop4
endi
sql drop stream if exists streams4;
sql drop stream if exists streams5;
sql drop database if exists test4;
sql drop database if exists test5;
sql drop database if exists result4;
sql drop database if exists result5;
print ===== step8
sql drop stream if exists streams8;
sql drop database if exists test8;
sql create database test8 vgroups 1;
sql use test8;
sql create table t1(ts timestamp, a int, b int , c int, d double);
sql create stream streams8 trigger at_once into streamt8 as select _wstart as ts, count(*) c1, count(d) c2, count(c) c3 from t1 partition by tbname interval(10s) ;
sql drop stream streams8;
sql create stream streams71 trigger at_once into streamt8(ts, c2) tags(group_id)as select _wstart, count(*) from t1 partition by tbname as group_id interval(10s);
sql insert into t1 values(1648791233000,1,2,3,1.0);
loop8:
sleep 300
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
sql select * from streamt8;
print $data00, $data01, $data02, $data03
print $data10, $data11, $data12, $data13
print $data20, $data21, $data22, $data23
if $rows != 1 then
print =====rows=$rows
goto loop8
endi
if $data01 != NULL then
print =====data01=$data01
goto loop8
endi
if $data02 != 1 then
print =====data02=$data02
goto loop8
endi
if $data03 != NULL then
print =====data03=$data03
goto loop8
endi
print ======over

5
tests/script/tsim/stream/udTableAndTag0.sim
View File

@ -39,7 +39,10 @@ sql select table_name from information_schema.ins_tables where db_name="result"
if $rows != 2 then
print =====rows=$rows
print $data00 $data10
print $data00
print $data10
print $data20
print $data30
goto loop0
endi

2
utils/tsim/src/simExe.c
View File

@ -31,7 +31,7 @@ void simLogSql(char *sql, bool useSharp) {
taosFprintfFile(pFile, "%s;\n", sql);
}
taosFsyncFile(pFile);
UNUSED(taosFsyncFile(pFile));
}
}