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

Merge branch '3.0' into fix/TD-20456

This commit is contained in:
Ganlin Zhao 2022-11-17 08:50:11 +08:00
parent a2f3eeb6f7 a186cc4491
commit 0b51e68c58
29 changed files with 289 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/taosws_CMakeLists.txt.in
View File

@ -2,7 +2,7 @@
# taosws-rs # taosws-rs
ExternalProject_Add(taosws-rs ExternalProject_Add(taosws-rs
GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git
GIT_TAG 7664c41 GIT_TAG 9843872
SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs" SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs"
BINARY_DIR "" BINARY_DIR ""
#BUILD_IN_SOURCE TRUE #BUILD_IN_SOURCE TRUE

3
docs/en/14-reference/13-schemaless/13-schemaless.md
View File

@ -8,6 +8,9 @@ will automatically add the required columns to ensure that the data written by t
The schemaless writing method creates super tables and their corresponding subtables. These are completely indistinguishable from the super tables and subtables created directly via SQL. You can write data directly to them via SQL statements. Note that the names of tables created by schemaless writing are based on fixed mapping rules for tag values, so they are not explicitly ideographic and they lack readability. The schemaless writing method creates super tables and their corresponding subtables. These are completely indistinguishable from the super tables and subtables created directly via SQL. You can write data directly to them via SQL statements. Note that the names of tables created by schemaless writing are based on fixed mapping rules for tag values, so they are not explicitly ideographic and they lack readability.
Tips:
The schemaless write will automatically create a table. You do not need to create a table manually, or an unknown error may occur.
## Schemaless Writing Line Protocol ## Schemaless Writing Line Protocol
TDengine's schemaless writing line protocol supports InfluxDB's Line Protocol, OpenTSDB's telnet line protocol, and OpenTSDB's JSON format protocol. However, when using these three protocols, you need to specify in the API the standard of the parsing protocol to be used for the input content. TDengine's schemaless writing line protocol supports InfluxDB's Line Protocol, OpenTSDB's telnet line protocol, and OpenTSDB's JSON format protocol. However, when using these three protocols, you need to specify in the API the standard of the parsing protocol to be used for the input content.

2
docs/zh/14-reference/13-schemaless/13-schemaless.md
View File

@ -8,6 +8,8 @@ description: 'Schemaless 写入方式，可以免于预先创建超级表/子表
无模式写入方式建立的超级表及其对应的子表与通过 SQL 直接建立的超级表和子表完全没有区别，你也可以通过，SQL 语句直接向其中写入数据。需要注意的是，通过无模式写入方式建立的表，其表名是基于标签值按照固定的映射规则生成，所以无法明确地进行表意，缺乏可读性。 无模式写入方式建立的超级表及其对应的子表与通过 SQL 直接建立的超级表和子表完全没有区别，你也可以通过，SQL 语句直接向其中写入数据。需要注意的是，通过无模式写入方式建立的表，其表名是基于标签值按照固定的映射规则生成，所以无法明确地进行表意，缺乏可读性。
注意：无模式写入会自动建表，不需要手动建表，手动建表的话可能会出现未知的错误。
## 无模式写入行协议 ## 无模式写入行协议
TDengine 的无模式写入的行协议兼容 InfluxDB 的 行协议（Line Protocol）、OpenTSDB 的 telnet 行协议、OpenTSDB 的 JSON 格式协议。但是使用这三种协议的时候，需要在 API 中指定输入内容使用解析协议的标准。 TDengine 的无模式写入的行协议兼容 InfluxDB 的 行协议（Line Protocol）、OpenTSDB 的 telnet 行协议、OpenTSDB 的 JSON 格式协议。但是使用这三种协议的时候，需要在 API 中指定输入内容使用解析协议的标准。

5
packaging/tools/windows_before_install.txt
View File

@ -1,3 +1,6 @@
TDengine is a high-efficient, scalable, high-available distributed time-series database, which makes a lot of optimizations on inserting and querying data, which is far more efficient than normal regular databases. So TDengine can meet the high requirements of IOT and other areas on storing and querying a large amount of data. TDengine is a high-efficient, scalable, high-available distributed time-series database, which makes a lot of optimizations on inserting and querying data, which is far more efficient than normal regular databases. So TDengine can meet the high requirements of IOT and other areas on storing and querying a large amount of data.
TDengine will be installed under C:\TDengine, users can modify configuration file C:\TDengine\cfg\taos.cfg, set the log file path or other parameters. TDengine will be installed under C:\TDengine, users can modify configuration file C:\TDengine\cfg\taos.cfg, set the log file path or other parameters.
To start/stop TDengine with administrator privileges: sc start/stop taosd
To start/stop taosAdapter with administrator privileges: sc start/stop taosadapter
Please manually remove C:\TDengine from your system PATH environment after you remove TDengine software.

2
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -287,7 +287,7 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) {
vmReleaseVnode(pMgmt, pVnode); vmReleaseVnode(pMgmt, pVnode);
} }
if (size < 0) { if (size < 0) {
dError("vgId:%d, can't get size from queue since %s, qtype:%d", vgId, terrstr(), qtype); dTrace("vgId:%d, can't get size from queue since %s, qtype:%d", vgId, terrstr(), qtype);
size = 0; size = 0;
} }
return size; return size;

1
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -508,6 +508,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
qDestroyQueryPlan(pPlan); qDestroyQueryPlan(pPlan);
return -1; return -1;
} }
sdbRelease(pSdb, pVgroup);
} }
} }
qDestroyQueryPlan(pPlan); qDestroyQueryPlan(pPlan);

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

@ -295,6 +295,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
return -1; return -1;
} }
pObj->sourceDbUid = pSourceDb->uid; pObj->sourceDbUid = pSourceDb->uid;
mndReleaseDb(pMnode, pSourceDb);
memcpy(pObj->targetSTbName, pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN); memcpy(pObj->targetSTbName, pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN);
@ -307,6 +308,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj,
pObj->targetStbUid = mndGenerateUid(pObj->targetSTbName, TSDB_TABLE_FNAME_LEN); pObj->targetStbUid = mndGenerateUid(pObj->targetSTbName, TSDB_TABLE_FNAME_LEN);
pObj->targetDbUid = pTargetDb->uid; pObj->targetDbUid = pTargetDb->uid;
mndReleaseDb(pMnode, pTargetDb);
pObj->sql = pCreate->sql; pObj->sql = pCreate->sql;
pObj->ast = pCreate->ast; pObj->ast = pCreate->ast;
@ -523,6 +525,7 @@ static int32_t mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStre
tFreeSMCreateStbReq(&createReq); tFreeSMCreateStbReq(&createReq);
mndFreeStb(&stbObj); mndFreeStb(&stbObj);
mndReleaseDb(pMnode, pDb);
return 0; return 0;
_OVER: _OVER:

4
source/dnode/mnode/impl/src/mndSubscribe.c
View File

@ -996,7 +996,7 @@ static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)&pConsumerEp->consumerId, false); colDataAppend(pColInfo, numOfRows, (const char *)&pConsumerEp->consumerId, false);
mDebug("mnd show subscrptions: topic %s, consumer %" PRId64 "cgroup %s vgid %d", varDataVal(topic), mDebug("mnd show subscriptions: topic %s, consumer %" PRId64 " cgroup %s vgid %d", varDataVal(topic),
pConsumerEp->consumerId, varDataVal(cgroup), pVgEp->vgId); pConsumerEp->consumerId, varDataVal(cgroup), pVgEp->vgId);
// offset // offset
@ -1044,7 +1044,7 @@ static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, NULL, true); colDataAppend(pColInfo, numOfRows, NULL, true);
mDebug("mnd show subscrptions(unassigned): topic %s, cgroup %s vgid %d", varDataVal(topic), varDataVal(cgroup), mDebug("mnd show subscriptions(unassigned): topic %s, cgroup %s vgid %d", varDataVal(topic), varDataVal(cgroup),
pVgEp->vgId); pVgEp->vgId);
// offset // offset

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

@ -88,6 +88,7 @@ typedef struct SBlockLoadSuppInfo {
int16_t* colIds; // column ids for loading file block data int16_t* colIds; // column ids for loading file block data
int32_t numOfCols; int32_t numOfCols;
char** buildBuf; // build string tmp buffer, todo remove it later after all string format being updated. char** buildBuf; // build string tmp buffer, todo remove it later after all string format being updated.
bool smaValid; // the sma on all queried columns are activated
} SBlockLoadSuppInfo; } SBlockLoadSuppInfo;
typedef struct SLastBlockReader { typedef struct SLastBlockReader {
@ -213,11 +214,10 @@ static bool hasDataInFileBlock(const SBlockData* pBlockData, const SFil
static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); } static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); }
static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) { static int32_t setColumnIdSlotList(SBlockLoadSuppInfo* pSupInfo, SSDataBlock* pBlock) {
SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
size_t numOfCols = blockDataGetNumOfCols(pBlock); size_t numOfCols = blockDataGetNumOfCols(pBlock);
pSupInfo->smaValid = true;
pSupInfo->numOfCols = numOfCols; pSupInfo->numOfCols = numOfCols;
pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t)); pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t));
pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES); pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES);
@ -239,6 +239,28 @@ static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static void updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) {
int32_t i = 0, j = 0;
while(i < pSchema->numOfCols && j < pSupInfo->numOfCols) {
STColumn* pTCol = &pSchema->columns[i];
if (pTCol->colId == pSupInfo->colIds[j]) {
if (!IS_BSMA_ON(pTCol)) {
pSupInfo->smaValid = false;
return;
}
i += 1;
j += 1;
} else if (pTCol->colId < pSupInfo->colIds[j]) {
// do nothing
i += 1;
} else {
ASSERT(0);
}
}
}
static int32_t initBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables) { static int32_t initBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables) {
int32_t num = numOfTables / pBuf->numPerBucket; int32_t num = numOfTables / pBuf->numPerBucket;
int32_t remainder = numOfTables % pBuf->numPerBucket; int32_t remainder = numOfTables % pBuf->numPerBucket;
@ -580,7 +602,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
// allocate buffer in order to load data blocks from file // allocate buffer in order to load data blocks from file
SBlockLoadSuppInfo* pSup = &pReader->suppInfo; SBlockLoadSuppInfo* pSup = &pReader->suppInfo;
pSup->pColAgg = taosArrayInit(4, sizeof(SColumnDataAgg)); pSup->pColAgg = taosArrayInit(pCond->numOfCols, sizeof(SColumnDataAgg));
pSup->plist = taosMemoryCalloc(pCond->numOfCols, POINTER_BYTES); pSup->plist = taosMemoryCalloc(pCond->numOfCols, POINTER_BYTES);
if (pSup->pColAgg == NULL || pSup->plist == NULL) { if (pSup->pColAgg == NULL || pSup->plist == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
@ -601,7 +623,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
goto _end; goto _end;
} }
setColumnIdSlotList(pReader, pReader->pResBlock); setColumnIdSlotList(&pReader->suppInfo, pReader->pResBlock);
*ppReader = pReader; *ppReader = pReader;
return code; return code;
@ -3763,6 +3785,10 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL
} }
} }
if (pReader->pSchema != NULL) {
updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo);
}
STsdbReader* p = pReader->innerReader[0] != NULL ? pReader->innerReader[0] : pReader; STsdbReader* p = pReader->innerReader[0] != NULL ? pReader->innerReader[0] : pReader;
pReader->status.pTableMap = createDataBlockScanInfo(p, pTableList, numOfTables); pReader->status.pTableMap = createDataBlockScanInfo(p, pTableList, numOfTables);
@ -4020,7 +4046,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS
} }
// there is no statistics data for composed block // there is no statistics data for composed block
if (pReader->status.composedDataBlock) { if (pReader->status.composedDataBlock || (!pReader->suppInfo.smaValid)) {
*pBlockStatis = NULL; *pBlockStatis = NULL;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -4060,7 +4086,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS
int32_t i = 0, j = 0; int32_t i = 0, j = 0;
size_t size = taosArrayGetSize(pSup->pColAgg); size_t size = taosArrayGetSize(pSup->pColAgg);
#if 0
while (j < numOfCols && i < size) { while (j < numOfCols && i < size) {
SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i); SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
if (pAgg->colId == pSup->colIds[j]) { if (pAgg->colId == pSup->colIds[j]) {
@ -4068,6 +4094,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS
pSup->plist[j] = pAgg; pSup->plist[j] = pAgg;
} else { } else {
*allHave = false; *allHave = false;
break;
} }
i += 1; i += 1;
j += 1; j += 1;
@ -4077,12 +4104,43 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS
j += 1; j += 1;
} }
} }
#else
// fill the all null data column
SArray* pNewAggList = taosArrayInit(numOfCols, sizeof(SColumnDataAgg));
while (j < numOfCols && i < size) {
SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
if (pAgg->colId == pSup->colIds[j]) {
taosArrayPush(pNewAggList, pAgg);
i += 1;
j += 1;
} else if (pAgg->colId < pSup->colIds[j]) {
i += 1;
} else if (pSup->colIds[j] < pAgg->colId) {
// all date in this block are null
SColumnDataAgg nullColAgg = {.colId = pSup->colIds[j], .numOfNull = pBlock->nRow};
taosArrayPush(pNewAggList, &nullColAgg);
j += 1;
}
}
taosArrayClear(pSup->pColAgg);
taosArrayAddAll(pSup->pColAgg, pNewAggList);
size_t num = taosArrayGetSize(pSup->pColAgg);
for(int32_t k = 0; k < num; ++k) {
pSup->plist[k] = taosArrayGet(pSup->pColAgg, k);
}
taosArrayDestroy(pNewAggList);
#endif
pReader->cost.smaDataLoad += 1; pReader->cost.smaDataLoad += 1;
*pBlockStatis = pSup->plist; *pBlockStatis = pSup->plist;
tsdbDebug("vgId:%d, succeed to load block SMA for uid %" PRIu64 ", %s", 0, pFBlock->uid, pReader->idStr); tsdbDebug("vgId:%d, succeed to load block SMA for uid %" PRIu64 ", %s", 0, pFBlock->uid, pReader->idStr);
return code; return code;
} }

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

@ -517,7 +517,7 @@ static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData,
for (int32_t iColData = 0; iColData < pBlockData->nColData; iColData++) { for (int32_t iColData = 0; iColData < pBlockData->nColData; iColData++) {
SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData); SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
if ((!pColData->smaOn) || IS_VAR_DATA_TYPE(pColData->type)) continue; if ((!pColData->smaOn) || IS_VAR_DATA_TYPE(pColData->type) || ((pColData->flag & HAS_VALUE) == 0)) continue;
SColumnDataAgg sma = {.colId = pColData->cid}; SColumnDataAgg sma = {.colId = pColData->cid};
tColDataCalcSMA[pColData->type](pColData, &sma.sum, &sma.max, &sma.min, &sma.numOfNull); tColDataCalcSMA[pColData->type](pColData, &sma.sum, &sma.max, &sma.min, &sma.numOfNull);

54
source/libs/executor/inc/executorimpl.h
View File

@ -154,8 +154,8 @@ typedef struct {
} SSchemaInfo; } SSchemaInfo;
typedef struct { typedef struct {
int32_t operatorType; int32_t operatorType;
int64_t refId; int64_t refId;
} SExchangeOpStopInfo; } SExchangeOpStopInfo;
typedef struct { typedef struct {
@ -261,10 +261,10 @@ typedef struct SLimitInfo {
} SLimitInfo; } SLimitInfo;
typedef struct SExchangeInfo { typedef struct SExchangeInfo {
SArray* pSources; SArray* pSources;
SArray* pSourceDataInfo; SArray* pSourceDataInfo;
tsem_t ready; tsem_t ready;
void* pTransporter; void* pTransporter;
// SArray<SSDataBlock*>, result block list, used to keep the multi-block that // SArray<SSDataBlock*>, result block list, used to keep the multi-block that
// passed by downstream operator // passed by downstream operator
SArray* pResultBlockList; SArray* pResultBlockList;
@ -275,7 +275,7 @@ typedef struct SExchangeInfo {
SLoadRemoteDataInfo loadInfo; SLoadRemoteDataInfo loadInfo;
uint64_t self; uint64_t self;
SLimitInfo limitInfo; SLimitInfo limitInfo;
int64_t openedTs; // start exec time stamp int64_t openedTs; // start exec time stamp
} SExchangeInfo; } SExchangeInfo;
typedef struct SScanInfo { typedef struct SScanInfo {
@ -310,9 +310,9 @@ typedef struct {
} SAggOptrPushDownInfo; } SAggOptrPushDownInfo;
typedef struct STableMetaCacheInfo { typedef struct STableMetaCacheInfo {
SLRUCache* pTableMetaEntryCache; // 100 by default SLRUCache* pTableMetaEntryCache; // 100 by default
uint64_t metaFetch; uint64_t metaFetch;
uint64_t cacheHit; uint64_t cacheHit;
} STableMetaCacheInfo; } STableMetaCacheInfo;
typedef struct STableScanInfo { typedef struct STableScanInfo {
@ -343,7 +343,7 @@ typedef struct STableMergeScanInfo {
int32_t tableEndIndex; int32_t tableEndIndex;
bool hasGroupId; bool hasGroupId;
uint64_t groupId; uint64_t groupId;
SArray* queryConds; // array of queryTableDataCond SArray* queryConds; // array of queryTableDataCond
STsdbReader* pReader; STsdbReader* pReader;
SReadHandle readHandle; SReadHandle readHandle;
int32_t bufPageSize; int32_t bufPageSize;
@ -358,7 +358,7 @@ typedef struct STableMergeScanInfo {
int64_t numOfRows; int64_t numOfRows;
SScanInfo scanInfo; SScanInfo scanInfo;
int32_t scanTimes; int32_t scanTimes;
SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context
SResultRowInfo* pResultRowInfo; SResultRowInfo* pResultRowInfo;
int32_t* rowEntryInfoOffset; int32_t* rowEntryInfoOffset;
SExprInfo* pExpr; SExprInfo* pExpr;
@ -504,6 +504,7 @@ typedef struct SStreamScanInfo {
STimeWindow updateWin; STimeWindow updateWin;
STimeWindowAggSupp twAggSup; STimeWindowAggSupp twAggSup;
SSDataBlock* pUpdateDataRes; SSDataBlock* pUpdateDataRes;
SHashObj* pGroupIdTbNameMap;
// status for tmq // status for tmq
SNodeList* pGroupTags; SNodeList* pGroupTags;
SNode* pTagCond; SNode* pTagCond;
@ -550,10 +551,10 @@ typedef struct SSysTableScanInfo {
} SSysTableScanInfo; } SSysTableScanInfo;
typedef struct SBlockDistInfo { typedef struct SBlockDistInfo {
SSDataBlock* pResBlock; SSDataBlock* pResBlock;
STsdbReader* pHandle; STsdbReader* pHandle;
SReadHandle readHandle; SReadHandle readHandle;
uint64_t uid; // table uid uint64_t uid; // table uid
} SBlockDistInfo; } SBlockDistInfo;
// todo remove this // todo remove this
@ -663,9 +664,9 @@ typedef struct SGroupbyOperatorInfo {
SAggSupporter aggSup; SAggSupporter aggSup;
SArray* pGroupCols; // group by columns, SArray<SColumn> SArray* pGroupCols; // group by columns, SArray<SColumn>
SArray* pGroupColVals; // current group column values, SArray<SGroupKeys> SArray* pGroupColVals; // current group column values, SArray<SGroupKeys>
bool isInit; // denote if current val is initialized or not bool isInit; // denote if current val is initialized or not
char* keyBuf; // group by keys for hash char* keyBuf; // group by keys for hash
int32_t groupKeyLen; // total group by column width int32_t groupKeyLen; // total group by column width
SGroupResInfo groupResInfo; SGroupResInfo groupResInfo;
SExprSupp scalarSup; SExprSupp scalarSup;
} SGroupbyOperatorInfo; } SGroupbyOperatorInfo;
@ -890,14 +891,14 @@ STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInter
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag); int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag);
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz); int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz);
void doDestroyExchangeOperatorInfo(void* param); void doDestroyExchangeOperatorInfo(void* param);
void setOperatorCompleted(SOperatorInfo* pOperator); void setOperatorCompleted(SOperatorInfo* pOperator);
void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status, void* pInfo, void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
SExecTaskInfo* pTaskInfo); void* pInfo, SExecTaskInfo* pTaskInfo);
void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo); void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
SSDataBlock* pBlock, int32_t rows, const char* idStr, STableMetaCacheInfo * pCache); int32_t rows, const char* idStr, STableMetaCacheInfo* pCache);
void cleanupAggSup(SAggSupporter* pAggSup); void cleanupAggSup(SAggSupporter* pAggSup);
void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle); void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle);
@ -992,7 +993,7 @@ void setTaskKilled(SExecTaskInfo* pTaskInfo);
void queryCostStatis(SExecTaskInfo* pTaskInfo); void queryCostStatis(SExecTaskInfo* pTaskInfo);
void doDestroyTask(SExecTaskInfo* pTaskInfo); void doDestroyTask(SExecTaskInfo* pTaskInfo);
void destroyOperatorInfo(SOperatorInfo* pOperator); void destroyOperatorInfo(SOperatorInfo* pOperator);
int32_t getMaximumIdleDurationSec(); int32_t getMaximumIdleDurationSec();
/* /*
@ -1038,6 +1039,7 @@ void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKE
uint64_t* pGp, void* pTbName); uint64_t* pGp, void* pTbName);
void printDataBlock(SSDataBlock* pBlock, const char* flag); void printDataBlock(SSDataBlock* pBlock, const char* flag);
uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId); uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup, int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
@ -1061,7 +1063,7 @@ int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResul
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult); int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult);
int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize); int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize);
void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order); void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order);
int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo *pInfo); int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/libs/executor/src/cachescanoperator.c
View File

@ -208,7 +208,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
if (pInfo->pseudoExprSup.numOfExprs > 0) { if (pInfo->pseudoExprSup.numOfExprs > 0) {
SExprSupp* pSup = &pInfo->pseudoExprSup; SExprSupp* pSup = &pInfo->pseudoExprSup;
STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pTableList)[0]; STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pList)[0];
pInfo->pRes->info.groupId = pKeyInfo->groupId; pInfo->pRes->info.groupId = pKeyInfo->groupId;
if (taosArrayGetSize(pInfo->pUidList) > 0) { if (taosArrayGetSize(pInfo->pUidList) > 0) {

36
source/libs/executor/src/scanoperator.c
View File

@ -1515,10 +1515,17 @@ static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pS
for (int32_t i = 0; i < pSrcBlock->info.rows; i++) { for (int32_t i = 0; i < pSrcBlock->info.rows; i++) {
uint64_t srcUid = srcUidData[i]; uint64_t srcUid = srcUidData[i];
uint64_t groupId = srcGp[i]; uint64_t groupId = srcGp[i];
char* tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0};
if (groupId == 0) { if (groupId == 0) {
groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version); groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version);
} }
appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId, NULL); if (pInfo->tbnameCalSup.pExprInfo) {
char* parTbname = taosHashGet(pInfo->pGroupIdTbNameMap, &groupId, sizeof(int64_t));
memcpy(varDataVal(tbname), parTbname, TSDB_TABLE_NAME_LEN);
varDataSetLen(tbname, strlen(varDataVal(tbname)));
}
appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId,
tbname[0] == 0 ? NULL : tbname);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -1562,9 +1569,16 @@ static void calBlockTag(SExprSupp* pTagCalSup, SSDataBlock* pBlock, SSDataBlock*
blockDataDestroy(pSrcBlock); blockDataDestroy(pSrcBlock);
} }
static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) { void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
SExprSupp* pTbNameCalSup = &pInfo->tbnameCalSup;
if (pTbNameCalSup == NULL || pTbNameCalSup->numOfExprs == 0) return; if (pTbNameCalSup == NULL || pTbNameCalSup->numOfExprs == 0) return;
if (pBlock == NULL || pBlock->info.rows == 0) return; if (pBlock == NULL || pBlock->info.rows == 0) return;
if (pBlock->info.groupId) {
char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
if (tbname != NULL) {
memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
}
}
SSDataBlock* pSrcBlock = blockCopyOneRow(pBlock, 0); SSDataBlock* pSrcBlock = blockCopyOneRow(pBlock, 0);
ASSERT(pSrcBlock->info.rows == 1); ASSERT(pSrcBlock->info.rows == 1);
@ -1592,6 +1606,11 @@ static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) {
pBlock->info.parTbName[0] = 0; pBlock->info.parTbName[0] = 0;
} }
if (pBlock->info.groupId) {
taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), pBlock->info.parTbName,
TSDB_TABLE_NAME_LEN);
}
blockDataDestroy(pSrcBlock); blockDataDestroy(pSrcBlock);
blockDataDestroy(pResBlock); blockDataDestroy(pResBlock);
} }
@ -1713,7 +1732,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex); blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
blockDataFreeRes((SSDataBlock*)pBlock); blockDataFreeRes((SSDataBlock*)pBlock);
calBlockTbName(&pInfo->tbnameCalSup, pInfo->pRes); calBlockTbName(pInfo, pInfo->pRes);
return 0; return 0;
} }
@ -1960,8 +1979,10 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) { if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) {
SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp); SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp);
if (pBlock != NULL) { if (pBlock != NULL) {
calBlockTbName(&pInfo->tbnameCalSup, pBlock); calBlockTbName(pInfo, pBlock);
updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); if (pInfo->pUpdateInfo) {
updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex);
}
qDebug("stream recover scan get block, rows %d", pBlock->info.rows); qDebug("stream recover scan get block, rows %d", pBlock->info.rows);
return pBlock; return pBlock;
} }
@ -2081,7 +2102,7 @@ FETCH_NEXT_BLOCK:
pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA; pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA;
checkUpdateData(pInfo, true, pSDB, false); checkUpdateData(pInfo, true, pSDB, false);
// printDataBlock(pSDB, "stream scan update"); // printDataBlock(pSDB, "stream scan update");
calBlockTbName(&pInfo->tbnameCalSup, pSDB); calBlockTbName(pInfo, pSDB);
return pSDB; return pSDB;
} }
blockDataCleanup(pInfo->pUpdateDataRes); blockDataCleanup(pInfo->pUpdateDataRes);
@ -2386,6 +2407,7 @@ static void destroyStreamScanOperatorInfo(void* param) {
} }
cleanupExprSupp(&pStreamScan->tbnameCalSup); cleanupExprSupp(&pStreamScan->tbnameCalSup);
taosHashCleanup(pStreamScan->pGroupIdTbNameMap);
updateInfoDestroy(pStreamScan->pUpdateInfo); updateInfoDestroy(pStreamScan->pUpdateInfo);
blockDataDestroy(pStreamScan->pRes); blockDataDestroy(pStreamScan->pRes);
@ -2443,6 +2465,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1) != 0) { if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1) != 0) {
goto _error; goto _error;
} }
pInfo->pGroupIdTbNameMap =
taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
} }
if (pTableScanNode->pTags != NULL) { if (pTableScanNode->pTags != NULL) {

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

@ -2924,6 +2924,8 @@ static void firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowInde
if (!pInfo->hasResult) { if (!pInfo->hasResult) {
pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock, NULL); pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock, NULL);
ASSERT(pCtx->subsidiaries.buf != NULL);
ASSERT(pCtx->subsidiaries.rowLen > 0);
} else { } else {
updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos); updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos);
} }
@ -4366,6 +4368,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
int32_t numOfParams = cJSON_GetArraySize(binDesc); int32_t numOfParams = cJSON_GetArraySize(binDesc);
int32_t startIndex; int32_t startIndex;
if (numOfParams != 4) { if (numOfParams != 4) {
cJSON_Delete(binDesc);
return false; return false;
} }
@ -4376,15 +4379,18 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
cJSON* infinity = cJSON_GetObjectItem(binDesc, "infinity"); cJSON* infinity = cJSON_GetObjectItem(binDesc, "infinity");
if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) {
cJSON_Delete(binDesc);
return false; return false;
} }
if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000
cJSON_Delete(binDesc);
return false; return false;
} }
if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) ||
(factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) {
cJSON_Delete(binDesc);
return false; return false;
} }
@ -4402,12 +4408,14 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
// linear bin process // linear bin process
if (width->valuedouble == 0) { if (width->valuedouble == 0) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
for (int i = 0; i < counter + 1; ++i) { for (int i = 0; i < counter + 1; ++i) {
intervals[startIndex] = start->valuedouble + i * width->valuedouble; intervals[startIndex] = start->valuedouble + i * width->valuedouble;
if (isinf(intervals[startIndex])) { if (isinf(intervals[startIndex])) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
startIndex++; startIndex++;
@ -4416,22 +4424,26 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
// log bin process // log bin process
if (start->valuedouble == 0) { if (start->valuedouble == 0) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
for (int i = 0; i < counter + 1; ++i) { for (int i = 0; i < counter + 1; ++i) {
intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0);
if (isinf(intervals[startIndex])) { if (isinf(intervals[startIndex])) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
startIndex++; startIndex++;
} }
} else { } else {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
@ -4446,6 +4458,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
} }
} else if (cJSON_IsArray(binDesc)) { /* user input bins */ } else if (cJSON_IsArray(binDesc)) { /* user input bins */
if (binType != USER_INPUT_BIN) { if (binType != USER_INPUT_BIN) {
cJSON_Delete(binDesc);
return false; return false;
} }
numOfBins = cJSON_GetArraySize(binDesc); numOfBins = cJSON_GetArraySize(binDesc);
@ -4453,6 +4466,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
cJSON* bin = binDesc->child; cJSON* bin = binDesc->child;
if (bin == NULL) { if (bin == NULL) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
int i = 0; int i = 0;
@ -4460,16 +4474,19 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
intervals[i] = bin->valuedouble; intervals[i] = bin->valuedouble;
if (!cJSON_IsNumber(bin)) { if (!cJSON_IsNumber(bin)) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
if (i != 0 && intervals[i] <= intervals[i - 1]) { if (i != 0 && intervals[i] <= intervals[i - 1]) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
bin = bin->next; bin = bin->next;
i++; i++;
} }
} else { } else {
cJSON_Delete(binDesc);
return false; return false;
} }
@ -4482,6 +4499,8 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t
} }
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return true; return true;
} }
@ -4495,18 +4514,23 @@ bool histogramFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultIn
pInfo->totalCount = 0; pInfo->totalCount = 0;
pInfo->normalized = 0; pInfo->normalized = 0;
int8_t binType = getHistogramBinType(varDataVal(pCtx->param[1].param.pz)); char *binTypeStr = strndup(varDataVal(pCtx->param[1].param.pz), varDataLen(pCtx->param[1].param.pz));
int8_t binType = getHistogramBinType(binTypeStr);
taosMemoryFree(binTypeStr);
if (binType == UNKNOWN_BIN) { if (binType == UNKNOWN_BIN) {
return false; return false;
} }
char* binDesc = varDataVal(pCtx->param[2].param.pz); char* binDesc = strndup(varDataVal(pCtx->param[2].param.pz), varDataLen(pCtx->param[2].param.pz));
int64_t normalized = pCtx->param[3].param.i; int64_t normalized = pCtx->param[3].param.i;
if (normalized != 0 && normalized != 1) { if (normalized != 0 && normalized != 1) {
return false; return false;
} }
if (!getHistogramBinDesc(pInfo, binDesc, binType, (bool)normalized)) { if (!getHistogramBinDesc(pInfo, binDesc, binType, (bool)normalized)) {
taosMemoryFree(binDesc);
return false; return false;
} }
taosMemoryFree(binDesc);
return true; return true;
} }

1
source/libs/planner/src/planSpliter.c
View File

@ -596,6 +596,7 @@ static int32_t stbSplSplitSessionForStream(SSplitContext* pCxt, SStableSplitInfo
int32_t index = 0; int32_t index = 0;
int32_t code = stbSplAppendWEnd(pPartWin, &index); int32_t code = stbSplAppendWEnd(pPartWin, &index);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
nodesDestroyNode(pMergeWin->pTsEnd);
pMergeWin->pTsEnd = nodesCloneNode(nodesListGetNode(pPartWin->node.pTargets, index)); pMergeWin->pTsEnd = nodesCloneNode(nodesListGetNode(pPartWin->node.pTargets, index));
if (NULL == pMergeWin->pTsEnd) { if (NULL == pMergeWin->pTsEnd) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;

29
source/libs/scalar/src/sclfunc.c
View File

@ -2620,6 +2620,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
int32_t numOfParams = cJSON_GetArraySize(binDesc); int32_t numOfParams = cJSON_GetArraySize(binDesc);
int32_t startIndex; int32_t startIndex;
if (numOfParams != 4) { if (numOfParams != 4) {
cJSON_Delete(binDesc);
return false; return false;
} }
@ -2630,15 +2631,18 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
cJSON *infinity = cJSON_GetObjectItem(binDesc, "infinity"); cJSON *infinity = cJSON_GetObjectItem(binDesc, "infinity");
if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) {
cJSON_Delete(binDesc);
return false; return false;
} }
if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000
cJSON_Delete(binDesc);
return false; return false;
} }
if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) ||
(factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) {
cJSON_Delete(binDesc);
return false; return false;
} }
@ -2656,12 +2660,14 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
// linear bin process // linear bin process
if (width->valuedouble == 0) { if (width->valuedouble == 0) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
for (int i = 0; i < counter + 1; ++i) { for (int i = 0; i < counter + 1; ++i) {
intervals[startIndex] = start->valuedouble + i * width->valuedouble; intervals[startIndex] = start->valuedouble + i * width->valuedouble;
if (isinf(intervals[startIndex])) { if (isinf(intervals[startIndex])) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
startIndex++; startIndex++;
@ -2670,22 +2676,26 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
// log bin process // log bin process
if (start->valuedouble == 0) { if (start->valuedouble == 0) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
for (int i = 0; i < counter + 1; ++i) { for (int i = 0; i < counter + 1; ++i) {
intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0);
if (isinf(intervals[startIndex])) { if (isinf(intervals[startIndex])) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
startIndex++; startIndex++;
} }
} else { } else {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
@ -2700,6 +2710,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
} }
} else if (cJSON_IsArray(binDesc)) { /* user input bins */ } else if (cJSON_IsArray(binDesc)) { /* user input bins */
if (binType != USER_INPUT_BIN) { if (binType != USER_INPUT_BIN) {
cJSON_Delete(binDesc);
return false; return false;
} }
numOfBins = cJSON_GetArraySize(binDesc); numOfBins = cJSON_GetArraySize(binDesc);
@ -2707,6 +2718,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
cJSON *bin = binDesc->child; cJSON *bin = binDesc->child;
if (bin == NULL) { if (bin == NULL) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
int i = 0; int i = 0;
@ -2714,16 +2726,19 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
intervals[i] = bin->valuedouble; intervals[i] = bin->valuedouble;
if (!cJSON_IsNumber(bin)) { if (!cJSON_IsNumber(bin)) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
if (i != 0 && intervals[i] <= intervals[i - 1]) { if (i != 0 && intervals[i] <= intervals[i - 1]) {
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return false; return false;
} }
bin = bin->next; bin = bin->next;
i++; i++;
} }
} else { } else {
cJSON_Delete(binDesc);
return false; return false;
} }
@ -2735,8 +2750,9 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin
(*bins)[i].count = 0; (*bins)[i].count = 0;
} }
cJSON_Delete(binDesc);
taosMemoryFree(intervals); taosMemoryFree(intervals);
cJSON_Delete(binDesc);
return true; return true;
} }
@ -2748,14 +2764,19 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
int32_t numOfBins = 0; int32_t numOfBins = 0;
int32_t totalCount = 0; int32_t totalCount = 0;
int8_t binType = getHistogramBinType(varDataVal(pInput[1].columnData->pData)); char *binTypeStr = strndup(varDataVal(pInput[1].columnData->pData), varDataLen(pInput[1].columnData->pData));
char *binDesc = varDataVal(pInput[2].columnData->pData); int8_t binType = getHistogramBinType(binTypeStr);
taosMemoryFree(binTypeStr);
char *binDesc = strndup(varDataVal(pInput[2].columnData->pData), varDataLen(pInput[2].columnData->pData));
int64_t normalized = *(int64_t *)(pInput[3].columnData->pData); int64_t normalized = *(int64_t *)(pInput[3].columnData->pData);
int32_t type = GET_PARAM_TYPE(pInput); int32_t type = GET_PARAM_TYPE(pInput);
if (!getHistogramBinDesc(&bins, &numOfBins, binDesc, binType, (bool)normalized)) { if (!getHistogramBinDesc(&bins, &numOfBins, binDesc, binType, (bool)normalized)) {
taosMemoryFree(binDesc);
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }
taosMemoryFree(binDesc);
for (int32_t i = 0; i < pInput->numOfRows; ++i) { for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) { if (colDataIsNull_s(pInputData, i)) {
@ -2785,6 +2806,8 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
} }
} }
colInfoDataEnsureCapacity(pOutputData, numOfBins, false);
for (int32_t k = 0; k < numOfBins; ++k) { for (int32_t k = 0; k < numOfBins; ++k) {
int32_t len; int32_t len;
char buf[512] = {0}; char buf[512] = {0};

3
source/libs/stream/src/stream.c
View File

@ -51,6 +51,7 @@ void streamSchedByTimer(void* param, void* tmrId) {
SStreamTask* pTask = (void*)param; SStreamTask* pTask = (void*)param;
if (atomic_load_8(&pTask->taskStatus) == TASK_STATUS__DROPPING) { if (atomic_load_8(&pTask->taskStatus) == TASK_STATUS__DROPPING) {
streamMetaReleaseTask(NULL, pTask);
return; return;
} }
@ -80,6 +81,8 @@ void streamSchedByTimer(void* param, void* tmrId) {
int32_t streamSetupTrigger(SStreamTask* pTask) { int32_t streamSetupTrigger(SStreamTask* pTask) {
if (pTask->triggerParam != 0) { if (pTask->triggerParam != 0) {
int32_t ref = atomic_add_fetch_32(&pTask->refCnt, 1);
ASSERT(ref == 2);
pTask->timer = taosTmrStart(streamSchedByTimer, (int32_t)pTask->triggerParam, pTask, streamEnv.timer); pTask->timer = taosTmrStart(streamSchedByTimer, (int32_t)pTask->triggerParam, pTask, streamEnv.timer);
pTask->triggerStatus = TASK_TRIGGER_STATUS__INACTIVE; pTask->triggerStatus = TASK_TRIGGER_STATUS__INACTIVE;
} }

9
source/libs/stream/src/streamMeta.c
View File

@ -80,7 +80,12 @@ void streamMetaClose(SStreamMeta* pMeta) {
pIter = taosHashIterate(pMeta->pTasks, pIter); pIter = taosHashIterate(pMeta->pTasks, pIter);
if (pIter == NULL) break; if (pIter == NULL) break;
SStreamTask* pTask = *(SStreamTask**)pIter; SStreamTask* pTask = *(SStreamTask**)pIter;
if (pTask->timer) {
taosTmrStop(pTask->timer);
pTask->timer = NULL;
}
tFreeSStreamTask(pTask); tFreeSStreamTask(pTask);
/*streamMetaReleaseTask(pMeta, pTask);*/
} }
taosHashCleanup(pMeta->pTasks); taosHashCleanup(pMeta->pTasks);
taosMemoryFree(pMeta->path); taosMemoryFree(pMeta->path);
@ -202,6 +207,10 @@ void streamMetaRemoveTask1(SStreamMeta* pMeta, int32_t taskId) {
if (ppTask) { if (ppTask) {
SStreamTask* pTask = *ppTask; SStreamTask* pTask = *ppTask;
taosHashRemove(pMeta->pTasks, &taskId, sizeof(int32_t)); taosHashRemove(pMeta->pTasks, &taskId, sizeof(int32_t));
/*if (pTask->timer) {
* taosTmrStop(pTask->timer);*/
/*pTask->timer = NULL;*/
/*}*/
atomic_store_8(&pTask->taskStatus, TASK_STATUS__DROPPING); atomic_store_8(&pTask->taskStatus, TASK_STATUS__DROPPING);
taosWLockLatch(&pMeta->lock); taosWLockLatch(&pMeta->lock);

6
source/libs/sync/inc/syncInt.h
View File

@ -73,11 +73,12 @@ typedef struct SSyncTimer {
SSyncHbTimerData hbData; SSyncHbTimerData hbData;
} SSyncTimer; } SSyncTimer;
typedef struct SElectTimer { typedef struct SElectTimerParam {
uint64_t logicClock; uint64_t logicClock;
SSyncNode* pSyncNode; SSyncNode* pSyncNode;
int64_t executeTime;
void* pData; void* pData;
} SElectTimer; } SElectTimerParam;
typedef struct SPeerState { typedef struct SPeerState {
SyncIndex lastSendIndex; SyncIndex lastSendIndex;
@ -153,6 +154,7 @@ typedef struct SSyncNode {
uint64_t electTimerLogicClock; uint64_t electTimerLogicClock;
TAOS_TMR_CALLBACK FpElectTimerCB; // Timer Fp TAOS_TMR_CALLBACK FpElectTimerCB; // Timer Fp
uint64_t electTimerCounter; uint64_t electTimerCounter;
SElectTimerParam electTimerParam;
// heartbeat timer // heartbeat timer
tmr_h pHeartbeatTimer; tmr_h pHeartbeatTimer;

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

@ -83,10 +83,14 @@ int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
ASSERT(pState != NULL); ASSERT(pState != NULL);
if (pMsg->lastSendIndex == pState->lastSendIndex) { if (pMsg->lastSendIndex == pState->lastSendIndex) {
int64_t timeNow = taosGetTimestampMs();
int64_t elapsed = timeNow - pState->lastSendTime;
sNTrace(ths, "sync-append-entries rtt elapsed:%" PRId64 ", index:%" PRId64, elapsed, pState->lastSendIndex);
syncNodeReplicateOne(ths, &(pMsg->srcId), true); syncNodeReplicateOne(ths, &(pMsg->srcId), true);
} }
} }
syncLogRecvAppendEntriesReply(ths, pMsg, "process"); syncLogRecvAppendEntriesReply(ths, pMsg, "process");
return 0; return 0;
} }

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

@ -429,7 +429,7 @@ bool syncIsReadyForRead(int64_t rid) {
int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) { int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) {
if (pSyncNode->peersNum == 0) { if (pSyncNode->peersNum == 0) {
sDebug("only one replica, cannot leader transfer"); sDebug("vgId:%d, only one replica, cannot leader transfer", pSyncNode->vgId);
terrno = TSDB_CODE_SYN_ONE_REPLICA; terrno = TSDB_CODE_SYN_ONE_REPLICA;
return -1; return -1;
} }
@ -445,7 +445,7 @@ int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) {
int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) { int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) {
if (pSyncNode->replicaNum == 1) { if (pSyncNode->replicaNum == 1) {
sDebug("only one replica, cannot leader transfer"); sDebug("vgId:%d, only one replica, cannot leader transfer", pSyncNode->vgId);
terrno = TSDB_CODE_SYN_ONE_REPLICA; terrno = TSDB_CODE_SYN_ONE_REPLICA;
return -1; return -1;
} }
@ -1104,8 +1104,16 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
int32_t ret = 0; int32_t ret = 0;
if (syncIsInit()) { if (syncIsInit()) {
pSyncNode->electTimerMS = ms; pSyncNode->electTimerMS = ms;
int64_t execTime = taosGetTimestampMs() + ms;
atomic_store_64(&(pSyncNode->electTimerParam.executeTime), execTime);
atomic_store_64(&(pSyncNode->electTimerParam.logicClock), pSyncNode->electTimerLogicClock);
pSyncNode->electTimerParam.pSyncNode = pSyncNode;
pSyncNode->electTimerParam.pData = NULL;
taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager, taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager,
&pSyncNode->pElectTimer); &pSyncNode->pElectTimer);
} else { } else {
sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId); sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId);
} }
@ -1855,27 +1863,35 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
} }
static void syncNodeEqElectTimer(void* param, void* tmrId) { static void syncNodeEqElectTimer(void* param, void* tmrId) {
SSyncNode* pNode = param;
if (!syncIsInit()) return; if (!syncIsInit()) return;
SSyncNode* pNode = (SSyncNode*)param;
if (pNode == NULL) return; if (pNode == NULL) return;
if (pNode->syncEqMsg == NULL) return; if (pNode->syncEqMsg == NULL) return;
int64_t tsNow = taosGetTimestampMs();
if (tsNow < pNode->electTimerParam.executeTime) return;
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
int32_t code = int32_t code =
syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerLogicClock, pNode->electTimerMS, pNode); syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerParam.logicClock, pNode->electTimerMS, pNode);
if (code != 0) { if (code != 0) {
sError("failed to build elect msg"); sError("failed to build elect msg");
return; return;
} }
SyncTimeout* pTimeout = rpcMsg.pCont; SyncTimeout* pTimeout = rpcMsg.pCont;
sTrace("enqueue elect msg lc:%" PRId64, pTimeout->logicClock); sNTrace(pNode, "enqueue elect msg lc:%" PRId64, pTimeout->logicClock);
code = pNode->syncEqMsg(pNode->msgcb, &rpcMsg); code = pNode->syncEqMsg(pNode->msgcb, &rpcMsg);
if (code != 0) { if (code != 0) {
sError("failed to sync enqueue elect msg since %s", terrstr()); sError("failed to sync enqueue elect msg since %s", terrstr());
rpcFreeCont(rpcMsg.pCont); rpcFreeCont(rpcMsg.pCont);
return;
} }
} }
@ -2059,6 +2075,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont; SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont;
pSyncMsg->cmd = SYNC_LOCAL_CMD_FOLLOWER_CMT; pSyncMsg->cmd = SYNC_LOCAL_CMD_FOLLOWER_CMT;
pSyncMsg->fcIndex = pMsg->commitIndex; pSyncMsg->fcIndex = pMsg->commitIndex;
SyncIndex fcIndex = pSyncMsg->fcIndex;
if (ths->syncEqMsg != NULL && ths->msgcb != NULL) { if (ths->syncEqMsg != NULL && ths->msgcb != NULL) {
int32_t code = ths->syncEqMsg(ths->msgcb, &rpcMsgLocalCmd); int32_t code = ths->syncEqMsg(ths->msgcb, &rpcMsgLocalCmd);
@ -2066,7 +2083,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
sError("vgId:%d, sync enqueue fc-commit msg error, code:%d", ths->vgId, code); sError("vgId:%d, sync enqueue fc-commit msg error, code:%d", ths->vgId, code);
rpcFreeCont(rpcMsgLocalCmd.pCont); rpcFreeCont(rpcMsgLocalCmd.pCont);
} else { } else {
sTrace("vgId:%d, sync enqueue fc-commit msg, fc-index:%" PRId64, ths->vgId, pSyncMsg->fcIndex); sTrace("vgId:%d, sync enqueue fc-commit msg, fc-index:%" PRId64, ths->vgId, fcIndex);
} }
} }
} }

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

@ -192,7 +192,9 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
// pMsg->privateTerm = pSender->privateTerm; // pMsg->privateTerm = pSender->privateTerm;
memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); if (pSender->pCurrentBlock != NULL) {
memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen);
}
// send msg // send msg
syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg);

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

@ -193,7 +193,7 @@ static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
} }
void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) { void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) {
if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; if (pNode == NULL || pNode->pRaftCfg == NULL || pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
int64_t currentTerm = pNode->pRaftStore->currentTerm; int64_t currentTerm = pNode->pRaftStore->currentTerm;
// save error code, otherwise it will be overwritten // save error code, otherwise it will be overwritten
@ -252,7 +252,7 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender, void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender,
const char* format, ...) { const char* format, ...) {
SSyncNode* pNode = pSender->pSyncNode; SSyncNode* pNode = pSender->pSyncNode;
if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; if (pNode == NULL || pNode->pRaftCfg == NULL || pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0}; SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) { if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) {
@ -304,7 +304,7 @@ void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dfla
void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotReceiver* pReceiver, void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotReceiver* pReceiver,
const char* format, ...) { const char* format, ...) {
SSyncNode* pNode = pReceiver->pSyncNode; SSyncNode* pNode = pReceiver->pSyncNode;
if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; if (pNode == NULL || pNode->pRaftCfg == NULL || pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0}; SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) { if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) {
@ -554,4 +554,4 @@ void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteRepl
syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port); syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode, "send sync-request-vote-reply to %s:%d {term:%" PRId64 ", grant:%d}, %s", host, port, pMsg->term, sNTrace(pSyncNode, "send sync-request-vote-reply to %s:%d {term:%" PRId64 ", grant:%d}, %s", host, port, pMsg->term,
pMsg->voteGranted, s); pMsg->voteGranted, s);
} }

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

@ -102,6 +102,7 @@ int32_t walCommit(SWal *pWal, int64_t ver) {
int32_t walRollback(SWal *pWal, int64_t ver) { int32_t walRollback(SWal *pWal, int64_t ver) {
taosThreadMutexLock(&pWal->mutex); taosThreadMutexLock(&pWal->mutex);
wInfo("vgId:%d, wal rollback for version %" PRId64, pWal->cfg.vgId, ver);
int64_t code; int64_t code;
char fnameStr[WAL_FILE_LEN]; char fnameStr[WAL_FILE_LEN];
if (ver > pWal->vers.lastVer || ver < pWal->vers.commitVer || ver <= pWal->vers.snapshotVer) { if (ver > pWal->vers.lastVer || ver < pWal->vers.commitVer || ver <= pWal->vers.snapshotVer) {
@ -123,8 +124,10 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
int fileSetSize = taosArrayGetSize(pWal->fileInfoSet); int fileSetSize = taosArrayGetSize(pWal->fileInfoSet);
for (int i = pWal->writeCur + 1; i < fileSetSize; i++) { for (int i = pWal->writeCur + 1; i < fileSetSize; i++) {
walBuildLogName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr); walBuildLogName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr);
wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr);
taosRemoveFile(fnameStr); taosRemoveFile(fnameStr);
walBuildIdxName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr); walBuildIdxName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr);
wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr);
taosRemoveFile(fnameStr); taosRemoveFile(fnameStr);
} }
// pop from fileInfoSet // pop from fileInfoSet
@ -157,6 +160,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr); walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND); TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
wDebug("vgId:%d, wal truncate file %s", pWal->cfg.vgId, fnameStr);
if (pLogFile == NULL) { if (pLogFile == NULL) {
// TODO // TODO
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
@ -321,12 +325,12 @@ int32_t walEndSnapshot(SWal *pWal) {
SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE); SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE);
if (pInfo) { if (pInfo) {
if (ver >= pInfo->lastVer) { if (ver >= pInfo->lastVer) {
pInfo++; pInfo--;
} }
if (POINTER_DISTANCE(pInfo, pWal->fileInfoSet->pData) > 0) { if (POINTER_DISTANCE(pInfo, pWal->fileInfoSet->pData) > 0) {
wDebug("vgId:%d, begin remove from %" PRId64, pWal->cfg.vgId, pInfo->firstVer); wDebug("vgId:%d, wal end remove for %" PRId64, pWal->cfg.vgId, pInfo->firstVer);
} else { } else {
wDebug("vgId:%d, no remove", pWal->cfg.vgId); wDebug("vgId:%d, wal no remove", pWal->cfg.vgId);
} }
// iterate files, until the searched result // iterate files, until the searched result
for (SWalFileInfo *iter = pWal->fileInfoSet->pData; iter < pInfo; iter++) { for (SWalFileInfo *iter = pWal->fileInfoSet->pData; iter < pInfo; iter++) {
@ -343,12 +347,12 @@ int32_t walEndSnapshot(SWal *pWal) {
for (int i = 0; i < deleteCnt; i++) { for (int i = 0; i < deleteCnt; i++) {
pInfo = taosArrayGet(pWal->fileInfoSet, i); pInfo = taosArrayGet(pWal->fileInfoSet, i);
walBuildLogName(pWal, pInfo->firstVer, fnameStr); walBuildLogName(pWal, pInfo->firstVer, fnameStr);
wDebug("vgId:%d, remove file %s", pWal->cfg.vgId, fnameStr); wDebug("vgId:%d, wal remove file %s", pWal->cfg.vgId, fnameStr);
if (taosRemoveFile(fnameStr) < 0) { if (taosRemoveFile(fnameStr) < 0) {
goto UPDATE_META; goto UPDATE_META;
} }
walBuildIdxName(pWal, pInfo->firstVer, fnameStr); walBuildIdxName(pWal, pInfo->firstVer, fnameStr);
wDebug("vgId:%d, remove file %s", pWal->cfg.vgId, fnameStr); wDebug("vgId:%d, wal remove file %s", pWal->cfg.vgId, fnameStr);
if (taosRemoveFile(fnameStr) < 0) { if (taosRemoveFile(fnameStr) < 0) {
ASSERT(0); ASSERT(0);
} }

22
tests/parallel_test/cases.task
View File

@ -115,11 +115,11 @@
,,y,script,./test.sh -f tsim/parser/interp.sim ,,y,script,./test.sh -f tsim/parser/interp.sim
#,,y,script,./test.sh -f tsim/parser/limit2.sim #,,y,script,./test.sh -f tsim/parser/limit2.sim
,,y,script,./test.sh -f tsim/parser/fourArithmetic-basic.sim ,,y,script,./test.sh -f tsim/parser/fourArithmetic-basic.sim
,,,script,./test.sh -f tsim/parser/function.sim ,,y,script,./test.sh -f tsim/parser/function.sim
,,y,script,./test.sh -f tsim/parser/groupby-basic.sim ,,y,script,./test.sh -f tsim/parser/groupby-basic.sim
,,,script,./test.sh -f tsim/parser/groupby.sim ,,y,script,./test.sh -f tsim/parser/groupby.sim
,,,script,./test.sh -f tsim/parser/having_child.sim ,,y,script,./test.sh -f tsim/parser/having_child.sim
,,,script,./test.sh -f tsim/parser/having.sim ,,y,script,./test.sh -f tsim/parser/having.sim
,,y,script,./test.sh -f tsim/parser/import_commit1.sim ,,y,script,./test.sh -f tsim/parser/import_commit1.sim
,,y,script,./test.sh -f tsim/parser/import_commit2.sim ,,y,script,./test.sh -f tsim/parser/import_commit2.sim
,,y,script,./test.sh -f tsim/parser/import_commit3.sim ,,y,script,./test.sh -f tsim/parser/import_commit3.sim
@ -140,7 +140,7 @@
,,,script,./test.sh -f tsim/parser/limit1.sim ,,,script,./test.sh -f tsim/parser/limit1.sim
,,y,script,./test.sh -f tsim/parser/mixed_blocks.sim ,,y,script,./test.sh -f tsim/parser/mixed_blocks.sim
,,y,script,./test.sh -f tsim/parser/nchar.sim ,,y,script,./test.sh -f tsim/parser/nchar.sim
,,,script,./test.sh -f tsim/parser/nestquery.sim ,,y,script,./test.sh -f tsim/parser/nestquery.sim
,,y,script,./test.sh -f tsim/parser/null_char.sim ,,y,script,./test.sh -f tsim/parser/null_char.sim
,,y,script,./test.sh -f tsim/parser/precision_ns.sim ,,y,script,./test.sh -f tsim/parser/precision_ns.sim
,,y,script,./test.sh -f tsim/parser/projection_limit_offset.sim ,,y,script,./test.sh -f tsim/parser/projection_limit_offset.sim
@ -173,7 +173,7 @@
,,y,script,./test.sh -f tsim/query/scalarFunction.sim ,,y,script,./test.sh -f tsim/query/scalarFunction.sim
,,y,script,./test.sh -f tsim/query/scalarNull.sim ,,y,script,./test.sh -f tsim/query/scalarNull.sim
,,y,script,./test.sh -f tsim/query/session.sim ,,y,script,./test.sh -f tsim/query/session.sim
,,,script,./test.sh -f tsim/query/udf.sim ,,y,script,./test.sh -f tsim/query/udf.sim
,,y,script,./test.sh -f tsim/qnode/basic1.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/snode/basic1.sim
,,,script,./test.sh -f tsim/mnode/basic1.sim ,,,script,./test.sh -f tsim/mnode/basic1.sim
@ -215,12 +215,12 @@
,,y,script,./test.sh -f tsim/stream/basic2.sim ,,y,script,./test.sh -f tsim/stream/basic2.sim
,,,script,./test.sh -f tsim/stream/drop_stream.sim ,,,script,./test.sh -f tsim/stream/drop_stream.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic1.sim ,,y,script,./test.sh -f tsim/stream/fillHistoryBasic1.sim
,,,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim ,,y,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim ,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim
,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim ,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim ,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
,,,script,./test.sh -f tsim/stream/distributeSession0.sim ,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,,script,./test.sh -f tsim/stream/session0.sim ,,y,script,./test.sh -f tsim/stream/session0.sim
,,y,script,./test.sh -f tsim/stream/session1.sim ,,y,script,./test.sh -f tsim/stream/session1.sim
,,y,script,./test.sh -f tsim/stream/state0.sim ,,y,script,./test.sh -f tsim/stream/state0.sim
,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim ,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
@ -232,10 +232,10 @@
,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim ,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
,,y,script,./test.sh -f tsim/stream/sliding.sim ,,y,script,./test.sh -f tsim/stream/sliding.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim ,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
,,,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim ,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim ,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
,,y,script,./test.sh -f tsim/stream/deleteInterval.sim ,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
,,,script,./test.sh -f tsim/stream/deleteSession.sim ,,y,script,./test.sh -f tsim/stream/deleteSession.sim
,,y,script,./test.sh -f tsim/stream/deleteState.sim ,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete0.sim ,,y,script,./test.sh -f tsim/stream/fillIntervalDelete0.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete1.sim ,,y,script,./test.sh -f tsim/stream/fillIntervalDelete1.sim

13
tests/script/test.sh
View File

@ -7,6 +7,7 @@
################################################## ##################################################
set +e set +e
#set -x
FILE_NAME= FILE_NAME=
RELEASE=0 RELEASE=0
@ -137,8 +138,16 @@ if [ -n "$FILE_NAME" ]; then
$PROGRAM -c $CFG_DIR -f $FILE_NAME -v $PROGRAM -c $CFG_DIR -f $FILE_NAME -v
else else
echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f $FILE_NAME echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f $FILE_NAME
$PROGRAM -c $CFG_DIR -f $FILE_NAME 2> $ASAN_DIR/tsim.asan echo "AsanDir:" $ASAN_DIR/tsim.asan
$CODE_DIR/sh/checkAsan.sh eval $PROGRAM -c $CFG_DIR -f $FILE_NAME 2> $ASAN_DIR/tsim.asan
result=$?
echo "Execute result: " $result
if [ $result -eq 0 ]; then
$CODE_DIR/sh/checkAsan.sh
else
exit 1
fi
fi fi
else else
echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f basicSuite.sim echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f basicSuite.sim

2
tests/script/tsim/compute/diff2.sim
View File

@ -83,7 +83,7 @@ sql select diff(c1), diff(c2) from $tb
sql select 2+diff(c1) from $tb sql select 2+diff(c1) from $tb
sql select diff(c1+2) from $tb sql select diff(c1+2) from $tb
sql_error select diff(c1) from $tb where ts > 0 and ts < now + 100m interval(10m) sql_error select diff(c1) from $tb where ts > 0 and ts < now + 100m interval(10m)
sql select diff(c1) from $mt #sql select diff(c1) from $mt
sql_error select diff(diff(c1)) from $tb sql_error select diff(diff(c1)) from $tb
sql_error select diff(c1) from m_di_tb1 where c2 like '2%' sql_error select diff(c1) from m_di_tb1 where c2 like '2%'

14
tests/script/tsim/parser/nestquery.sim
View File

@ -338,13 +338,13 @@ if $data03 != @20-09-15 00:00:00.000@ then
return -1 return -1
endi endi
sql select diff(val) from (select c1 val from nest_tb0); #sql select diff(val) from (select c1 val from nest_tb0);
if $rows != 9999 then #if $rows != 9999 then
return -1 # return -1
endi #endi
if $data00 != 1 then #if $data00 != 1 then
return -1 # return -1
endi #endi
sql_error select last_row(*) from (select * from nest_tb0) having c1 > 0 sql_error select last_row(*) from (select * from nest_tb0) having c1 > 0

4
tests/script/tsim/valgrind/checkError6.sim
View File

@ -92,7 +92,7 @@ sql select avg(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol
sql show table distributed stb sql show table distributed stb
sql select count(1) from stb sql select count(1) from stb
sql select count(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m) sql select count(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m)
sql select diff(tbcol) from stb where ts <= 1601481840000 #sql select diff(tbcol) from stb where ts <= 1601481840000
sql select first(tbcol), last(tbcol) as c from stb group by tgcol sql select first(tbcol), last(tbcol) as c from stb group by tgcol
sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 and tbcol2 is null partition by tgcol interval(1m) sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 and tbcol2 is null partition by tgcol interval(1m)
sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m) sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m)
@ -176,7 +176,7 @@ sql select avg(tbcol) as c from stb group by tgcol
sql select avg(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m) sql select avg(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m)
sql show table distributed stb sql show table distributed stb
sql select count(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m) sql select count(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m)
sql select diff(tbcol) from stb where ts <= 1601481840000 #sql select diff(tbcol) from stb where ts <= 1601481840000
sql select first(tbcol), last(tbcol) as c from stb group by tgcol sql select first(tbcol), last(tbcol) as c from stb group by tgcol
sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 and tbcol2 is null partition by tgcol interval(1m) sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 and tbcol2 is null partition by tgcol interval(1m)
sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m) sql select first(tbcol), last(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol interval(1m)