diff --git a/cmake/taosws_CMakeLists.txt.in b/cmake/taosws_CMakeLists.txt.in index 5c448e2c1d88f31b6a644f7662ce59cb80a5333a..7d48eb9d8ab0eceee62a7fb641d05e2f6b52ec83 100644 --- a/cmake/taosws_CMakeLists.txt.in +++ b/cmake/taosws_CMakeLists.txt.in @@ -2,7 +2,7 @@ # taosws-rs ExternalProject_Add(taosws-rs GIT_REPOSITORY https://github.com/taosdata/taos-connector-rust.git - GIT_TAG 7664c41 + GIT_TAG 9843872 SOURCE_DIR "${TD_SOURCE_DIR}/tools/taosws-rs" BINARY_DIR "" #BUILD_IN_SOURCE TRUE diff --git a/docs/en/12-taos-sql/19-limit.md b/docs/en/12-taos-sql/19-limit.md index 678c38a22ea763187cd0c87dceae3bf6ca03957c..f00ec90f5755d3d004a35b210f93560c6f719536 100644 --- a/docs/en/12-taos-sql/19-limit.md +++ b/docs/en/12-taos-sql/19-limit.md @@ -23,7 +23,7 @@ The following characters cannot occur in a password: single quotation marks ('), ## General Limits -- Maximum length of database name is 32 bytes +- Maximum length of database name is 64 bytes - Maximum length of table name is 192 bytes, excluding the database name prefix and the separator. - Maximum length of each data row is 48K bytes. Note that the upper limit includes the extra 2 bytes consumed by each column of BINARY/NCHAR type. - The maximum length of a column name is 64 bytes. @@ -35,7 +35,7 @@ The following characters cannot occur in a password: single quotation marks ('), - Maximum numbers of databases, STables, tables are dependent only on the system resources. - The number of replicas can only be 1 or 3. - The maximum length of a username is 23 bytes. -- The maximum length of a password is 15 bytes. +- The maximum length of a password is 128 bytes. - The maximum number of rows depends on system resources. - The maximum number of vnodes in a database is 1024. diff --git a/docs/en/14-reference/13-schemaless/13-schemaless.md b/docs/en/14-reference/13-schemaless/13-schemaless.md index 5b7924ce56b240d34ab139f6160839a56438dc6b..10321ab083e6e654e66cb73f1bc21f9fbd678fda 100644 --- a/docs/en/14-reference/13-schemaless/13-schemaless.md +++ b/docs/en/14-reference/13-schemaless/13-schemaless.md @@ -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. +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 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. diff --git a/docs/zh/12-taos-sql/19-limit.md b/docs/zh/12-taos-sql/19-limit.md index a9743adddabe96440ffca8c8585787081d29398f..7b6692f1b733042fde3103de1e9fa06bf4dda542 100644 --- a/docs/zh/12-taos-sql/19-limit.md +++ b/docs/zh/12-taos-sql/19-limit.md @@ -24,19 +24,19 @@ description: 合法字符集和命名中的限制规则 ## 一般限制 -- 数据库名最大长度为 32 -- 表名最大长度为 192，不包括数据库名前缀和分隔符 +- 数据库名最大长度为 64 字节 +- 表名最大长度为 192 字节，不包括数据库名前缀和分隔符 - 每行数据最大长度 48KB （注意：数据行内每个 BINARY/NCHAR 类型的列还会额外占用 2 个字节的存储位置） -- 列名最大长度为 64 +- 列名最大长度为 64 字节 - 最多允许 4096 列，最少需要 2 列，第一列必须是时间戳。 -- 标签名最大长度为 64 +- 标签名最大长度为 64 字节 - 最多允许 128 个，至少要有 1 个标签，一个表中标签值的总长度不超过 16KB - SQL 语句最大长度 1048576 个字符 - SELECT 语句的查询结果，最多允许返回 4096 列（语句中的函数调用可能也会占用一些列空间），超限时需要显式指定较少的返回数据列，以避免语句执行报错 - 库的数目，超级表的数目、表的数目，系统不做限制，仅受系统资源限制 - 数据库的副本数只能设置为 1 或 3 -- 用户名的最大长度是 23 个字节 -- 用户密码的最大长度是 15 个字节 +- 用户名的最大长度是 23 字节 +- 用户密码的最大长度是 128 字节 - 总数据行数取决于可用资源 - 单个数据库的虚拟结点数上限为 1024 diff --git a/docs/zh/14-reference/13-schemaless/13-schemaless.md b/docs/zh/14-reference/13-schemaless/13-schemaless.md index b89f625376404677af21f0d48b77c3df5ebd748a..3aebd616a075394dbeddc9e608cd2a57bb8cf844 100644 --- a/docs/zh/14-reference/13-schemaless/13-schemaless.md +++ b/docs/zh/14-reference/13-schemaless/13-schemaless.md @@ -8,6 +8,8 @@ description: 'Schemaless 写入方式，可以免于预先创建超级表/子表 无模式写入方式建立的超级表及其对应的子表与通过 SQL 直接建立的超级表和子表完全没有区别，你也可以通过，SQL 语句直接向其中写入数据。需要注意的是，通过无模式写入方式建立的表，其表名是基于标签值按照固定的映射规则生成，所以无法明确地进行表意，缺乏可读性。 +注意：无模式写入会自动建表，不需要手动建表，手动建表的话可能会出现未知的错误。 + ## 无模式写入行协议 TDengine 的无模式写入的行协议兼容 InfluxDB 的 行协议（Line Protocol）、OpenTSDB 的 telnet 行协议、OpenTSDB 的 JSON 格式协议。但是使用这三种协议的时候，需要在 API 中指定输入内容使用解析协议的标准。 diff --git a/include/libs/stream/tstreamUpdate.h b/include/libs/stream/tstreamUpdate.h index 1c490852f92ed9244aeea173c5f2b33d58453326..ab328c6ad5c20170bf6b3c7ca7f4930b9082ad17 100644 --- a/include/libs/stream/tstreamUpdate.h +++ b/include/libs/stream/tstreamUpdate.h @@ -47,7 +47,7 @@ typedef struct SUpdateInfo { SUpdateInfo *updateInfoInitP(SInterval *pInterval, int64_t watermark); SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t watermark); -void updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t primaryTsCol); +TSKEY updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t primaryTsCol); bool updateInfoIsUpdated(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts); bool updateInfoIsTableInserted(SUpdateInfo *pInfo, int64_t tbUid); void updateInfoSetScanRange(SUpdateInfo *pInfo, STimeWindow *pWin, uint64_t groupId, uint64_t version); diff --git a/packaging/tools/windows_before_install.txt b/packaging/tools/windows_before_install.txt index b793a3e8014493b374c3b7bd79d7f535ca14f555..ef783bf10badedc733e87946a323d499b60fc5ac 100644 --- a/packaging/tools/windows_before_install.txt +++ b/packaging/tools/windows_before_install.txt @@ -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 will be installed under C:\TDengine, users can modify configuration file C:\TDengine\cfg\taos.cfg, set the log file path or other parameters. \ No newline at end of file +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. diff --git a/source/common/src/ttime.c b/source/common/src/ttime.c index 1b140e6c042783c31180a7f1d8cfc80312052fc7..a106a09a69a5a1d7d707c2d40548672c384a2987 100644 --- a/source/common/src/ttime.c +++ b/source/common/src/ttime.c @@ -436,21 +436,24 @@ int64_t convertTimePrecision(int64_t utime, int32_t fromPrecision, int32_t toPre ASSERT(toPrecision == TSDB_TIME_PRECISION_MILLI || toPrecision == TSDB_TIME_PRECISION_MICRO || toPrecision == TSDB_TIME_PRECISION_NANO); - double tempResult = (double)utime; - switch (fromPrecision) { case TSDB_TIME_PRECISION_MILLI: { switch (toPrecision) { case TSDB_TIME_PRECISION_MILLI: return utime; case TSDB_TIME_PRECISION_MICRO: - tempResult *= 1000; - utime *= 1000; - goto end_; + if (utime > INT64_MAX / 1000) { + return INT64_MAX; + } + return utime * 1000; case TSDB_TIME_PRECISION_NANO: - tempResult *= 1000000; - utime *= 1000000; - goto end_; + if (utime > INT64_MAX / 1000000) { + return INT64_MAX; + } + return utime * 1000000; + default: + ASSERT(0); + return utime; } } // end from milli case TSDB_TIME_PRECISION_MICRO: { @@ -460,9 +463,13 @@ int64_t convertTimePrecision(int64_t utime, int32_t fromPrecision, int32_t toPre case TSDB_TIME_PRECISION_MICRO: return utime; case TSDB_TIME_PRECISION_NANO: - tempResult *= 1000; - utime *= 1000; - goto end_; + if (utime > INT64_MAX / 1000) { + return INT64_MAX; + } + return utime * 1000; + default: + ASSERT(0); + return utime; } } // end from micro case TSDB_TIME_PRECISION_NANO: { @@ -473,17 +480,17 @@ int64_t convertTimePrecision(int64_t utime, int32_t fromPrecision, int32_t toPre return utime / 1000; case TSDB_TIME_PRECISION_NANO: return utime; + default: + ASSERT(0); + return utime; } } // end from nano default: { - assert(0); + ASSERT(0); return utime; // only to pass windows compilation } } // end switch fromPrecision -end_: - if (tempResult >= (double)INT64_MAX) return INT64_MAX; - if (tempResult <= (double)INT64_MIN) return INT64_MIN; // INT64_MIN means NULL return utime; } @@ -599,18 +606,33 @@ int32_t convertStringToTimestamp(int16_t type, char* inputData, int64_t timePrec static int32_t getDuration(int64_t val, char unit, int64_t* result, int32_t timePrecision) { switch (unit) { case 's': + if (val > INT64_MAX / MILLISECOND_PER_SECOND) { + return -1; + } (*result) = convertTimePrecision(val * MILLISECOND_PER_SECOND, TSDB_TIME_PRECISION_MILLI, timePrecision); break; case 'm': + if (val > INT64_MAX / MILLISECOND_PER_MINUTE) { + return -1; + } (*result) = convertTimePrecision(val * MILLISECOND_PER_MINUTE, TSDB_TIME_PRECISION_MILLI, timePrecision); break; case 'h': + if (val > INT64_MAX / MILLISECOND_PER_MINUTE) { + return -1; + } (*result) = convertTimePrecision(val * MILLISECOND_PER_HOUR, TSDB_TIME_PRECISION_MILLI, timePrecision); break; case 'd': + if (val > INT64_MAX / MILLISECOND_PER_DAY) { + return -1; + } (*result) = convertTimePrecision(val * MILLISECOND_PER_DAY, TSDB_TIME_PRECISION_MILLI, timePrecision); break; case 'w': + if (val > INT64_MAX / MILLISECOND_PER_WEEK) { + return -1; + } (*result) = convertTimePrecision(val * MILLISECOND_PER_WEEK, TSDB_TIME_PRECISION_MILLI, timePrecision); break; case 'a': @@ -650,7 +672,7 @@ int32_t parseAbsoluteDuration(const char* token, int32_t tokenlen, int64_t* dura /* get the basic numeric value */ int64_t timestamp = taosStr2Int64(token, &endPtr, 10); - if (errno != 0) { + if (timestamp < 0 || errno != 0) { return -1; } @@ -668,7 +690,7 @@ int32_t parseNatualDuration(const char* token, int32_t tokenLen, int64_t* durati /* get the basic numeric value */ *duration = taosStr2Int64(token, NULL, 10); - if (errno != 0) { + if (*duration < 0 || errno != 0) { return -1; } diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 17bc1526b5fd6e95df1b83f6aad3c31106339fd1..24240e82efc2a535b7da096888a63fd43f8ce3f7 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -287,7 +287,7 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { vmReleaseVnode(pMgmt, pVnode); } 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; } return size; diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 7e7c6ee0b60242b1d2a34e17e332a2e571031c5b..e1886511b714fa0a92b1b03d998730e606fa5042 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -508,6 +508,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) { qDestroyQueryPlan(pPlan); return -1; } + sdbRelease(pSdb, pVgroup); } } qDestroyQueryPlan(pPlan); diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 2cfb81a817a2f6de90aeacc0d12c14ac9edec011..62247f2c2a36592ef0983be69b8188a8f6980176 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -295,6 +295,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj, return -1; } pObj->sourceDbUid = pSourceDb->uid; + mndReleaseDb(pMnode, pSourceDb); 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->targetDbUid = pTargetDb->uid; + mndReleaseDb(pMnode, pTargetDb); pObj->sql = pCreate->sql; pObj->ast = pCreate->ast; @@ -523,6 +525,7 @@ static int32_t mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStre tFreeSMCreateStbReq(&createReq); mndFreeStb(&stbObj); + mndReleaseDb(pMnode, pDb); return 0; _OVER: diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index fd798461047b63f790079521159d53b8c928ed48..58c89d76aa41542d4138ed4b638333edc74ef947 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -996,7 +996,7 @@ static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); 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); // offset @@ -1044,7 +1044,7 @@ static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); 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); // offset diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c index e00d0d955ec820176cb696fa652c5e7fdd374edb..9d3b86c79bfb3be454611e6532763c5865991700 100644 --- a/source/dnode/mnode/impl/src/mndVgroup.c +++ b/source/dnode/mnode/impl/src/mndVgroup.c @@ -1907,6 +1907,7 @@ static int32_t mndBalanceVgroup(SMnode *pMnode, SRpcMsg *pReq, SArray *pArray) { } _OVER: + taosHashCleanup(pBalancedVgroups); mndTransDrop(pTrans); return code; } diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index af368d33e0234396f813539827515d90d71744ed..986cba8b17b17936d3dc0ded647824638a5f1101 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -88,6 +88,7 @@ typedef struct SBlockLoadSuppInfo { int16_t* colIds; // column ids for loading file block data int32_t numOfCols; 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; 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 int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) { - SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo; - +static int32_t setColumnIdSlotList(SBlockLoadSuppInfo* pSupInfo, SSDataBlock* pBlock) { size_t numOfCols = blockDataGetNumOfCols(pBlock); + pSupInfo->smaValid = true; pSupInfo->numOfCols = numOfCols; pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t)); pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES); @@ -239,6 +239,28 @@ static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) { 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) { int32_t num = 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 SBlockLoadSuppInfo* pSup = &pReader->suppInfo; - pSup->pColAgg = taosArrayInit(4, sizeof(SColumnDataAgg)); + pSup->pColAgg = taosArrayInit(pCond->numOfCols, sizeof(SColumnDataAgg)); pSup->plist = taosMemoryCalloc(pCond->numOfCols, POINTER_BYTES); if (pSup->pColAgg == NULL || pSup->plist == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; @@ -601,7 +623,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd goto _end; } - setColumnIdSlotList(pReader, pReader->pResBlock); + setColumnIdSlotList(&pReader->suppInfo, pReader->pResBlock); *ppReader = pReader; 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; 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 - if (pReader->status.composedDataBlock) { + if (pReader->status.composedDataBlock || (!pReader->suppInfo.smaValid)) { *pBlockStatis = NULL; return TSDB_CODE_SUCCESS; } @@ -4060,7 +4086,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS int32_t i = 0, j = 0; size_t size = taosArrayGetSize(pSup->pColAgg); - +#if 0 while (j < numOfCols && i < size) { SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i); if (pAgg->colId == pSup->colIds[j]) { @@ -4068,6 +4094,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS pSup->plist[j] = pAgg; } else { *allHave = false; + break; } i += 1; j += 1; @@ -4077,12 +4104,43 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SColumnDataAgg*** pBlockS 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; *pBlockStatis = pSup->plist; tsdbDebug("vgId:%d, succeed to load block SMA for uid %" PRIu64 ", %s", 0, pFBlock->uid, pReader->idStr); - return code; } diff --git a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c index 8c1b0004a38201c1c9d827070137d69de85f4f7a..294a4bd3e4c404ddb33a4e3fe7cd0c5fe2203a87 100644 --- a/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c +++ b/source/dnode/vnode/src/tsdb/tsdbReaderWriter.c @@ -517,7 +517,7 @@ static int32_t tsdbWriteBlockSma(SDataFWriter *pWriter, SBlockData *pBlockData, for (int32_t iColData = 0; iColData < pBlockData->nColData; 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}; tColDataCalcSMA[pColData->type](pColData, &sma.sum, &sma.max, &sma.min, &sma.numOfNull); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index b0da277cfb762bcfa3789a692db9e171b7c6336d..0e2635459d1a3249bc272769c1628be3095e0851 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -154,8 +154,8 @@ typedef struct { } SSchemaInfo; typedef struct { - int32_t operatorType; - int64_t refId; + int32_t operatorType; + int64_t refId; } SExchangeOpStopInfo; typedef struct { @@ -261,10 +261,10 @@ typedef struct SLimitInfo { } SLimitInfo; typedef struct SExchangeInfo { - SArray* pSources; - SArray* pSourceDataInfo; - tsem_t ready; - void* pTransporter; + SArray* pSources; + SArray* pSourceDataInfo; + tsem_t ready; + void* pTransporter; // SArray, result block list, used to keep the multi-block that // passed by downstream operator SArray* pResultBlockList; @@ -275,7 +275,7 @@ typedef struct SExchangeInfo { SLoadRemoteDataInfo loadInfo; uint64_t self; SLimitInfo limitInfo; - int64_t openedTs; // start exec time stamp + int64_t openedTs; // start exec time stamp } SExchangeInfo; typedef struct SScanInfo { @@ -310,9 +310,9 @@ typedef struct { } SAggOptrPushDownInfo; typedef struct STableMetaCacheInfo { - SLRUCache* pTableMetaEntryCache; // 100 by default - uint64_t metaFetch; - uint64_t cacheHit; + SLRUCache* pTableMetaEntryCache; // 100 by default + uint64_t metaFetch; + uint64_t cacheHit; } STableMetaCacheInfo; typedef struct STableScanInfo { @@ -343,7 +343,7 @@ typedef struct STableMergeScanInfo { int32_t tableEndIndex; bool hasGroupId; uint64_t groupId; - SArray* queryConds; // array of queryTableDataCond + SArray* queryConds; // array of queryTableDataCond STsdbReader* pReader; SReadHandle readHandle; int32_t bufPageSize; @@ -358,7 +358,7 @@ typedef struct STableMergeScanInfo { int64_t numOfRows; SScanInfo scanInfo; 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; int32_t* rowEntryInfoOffset; SExprInfo* pExpr; @@ -504,6 +504,7 @@ typedef struct SStreamScanInfo { STimeWindow updateWin; STimeWindowAggSupp twAggSup; SSDataBlock* pUpdateDataRes; + SHashObj* pGroupIdTbNameMap; // status for tmq SNodeList* pGroupTags; SNode* pTagCond; @@ -550,10 +551,10 @@ typedef struct SSysTableScanInfo { } SSysTableScanInfo; typedef struct SBlockDistInfo { - SSDataBlock* pResBlock; - STsdbReader* pHandle; - SReadHandle readHandle; - uint64_t uid; // table uid + SSDataBlock* pResBlock; + STsdbReader* pHandle; + SReadHandle readHandle; + uint64_t uid; // table uid } SBlockDistInfo; // todo remove this @@ -663,9 +664,9 @@ typedef struct SGroupbyOperatorInfo { SAggSupporter aggSup; SArray* pGroupCols; // group by columns, SArray SArray* pGroupColVals; // current group column values, SArray - bool isInit; // denote if current val is initialized or not - char* keyBuf; // group by keys for hash - int32_t groupKeyLen; // total group by column width + bool isInit; // denote if current val is initialized or not + char* keyBuf; // group by keys for hash + int32_t groupKeyLen; // total group by column width SGroupResInfo groupResInfo; SExprSupp scalarSup; } 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 getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz); -void doDestroyExchangeOperatorInfo(void* param); +void doDestroyExchangeOperatorInfo(void* param); void setOperatorCompleted(SOperatorInfo* pOperator); -void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status, void* pInfo, - SExecTaskInfo* pTaskInfo); +void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status, + void* pInfo, SExecTaskInfo* pTaskInfo); void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo); -int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, - SSDataBlock* pBlock, int32_t rows, const char* idStr, STableMetaCacheInfo * pCache); +int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock, + int32_t rows, const char* idStr, STableMetaCacheInfo* pCache); void cleanupAggSup(SAggSupporter* pAggSup); void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle); @@ -992,7 +993,7 @@ void setTaskKilled(SExecTaskInfo* pTaskInfo); void queryCostStatis(SExecTaskInfo* pTaskInfo); void doDestroyTask(SExecTaskInfo* pTaskInfo); -void destroyOperatorInfo(SOperatorInfo* pOperator); +void destroyOperatorInfo(SOperatorInfo* pOperator); int32_t getMaximumIdleDurationSec(); /* @@ -1038,6 +1039,7 @@ void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKE uint64_t* pGp, void* pTbName); void printDataBlock(SSDataBlock* pBlock, const char* flag); 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, 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 saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize); 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 } diff --git a/source/libs/executor/src/cachescanoperator.c b/source/libs/executor/src/cachescanoperator.c index 6f9084ce52c3ae2caac867c73f23f40b95bff95f..b78ba8ac0abe5d2a2fa951d179c535cfab1b3608 100644 --- a/source/libs/executor/src/cachescanoperator.c +++ b/source/libs/executor/src/cachescanoperator.c @@ -208,7 +208,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) { if (pInfo->pseudoExprSup.numOfExprs > 0) { SExprSupp* pSup = &pInfo->pseudoExprSup; - STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pTableList)[0]; + STableKeyInfo* pKeyInfo = &((STableKeyInfo*)pList)[0]; pInfo->pRes->info.groupId = pKeyInfo->groupId; if (taosArrayGetSize(pInfo->pUidList) > 0) { diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 77d2aeba282ce71b1ca03351119cb242f25235b3..05ccda26c69781bcff8d2e707c67a767bf1ad3dd 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1515,10 +1515,17 @@ static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pS for (int32_t i = 0; i < pSrcBlock->info.rows; i++) { uint64_t srcUid = srcUidData[i]; uint64_t groupId = srcGp[i]; + char* tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0}; if (groupId == 0) { 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; } @@ -1562,9 +1569,16 @@ static void calBlockTag(SExprSupp* pTagCalSup, SSDataBlock* pBlock, SSDataBlock* 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 (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); ASSERT(pSrcBlock->info.rows == 1); @@ -1592,6 +1606,11 @@ static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) { 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(pResBlock); } @@ -1713,7 +1732,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex); blockDataFreeRes((SSDataBlock*)pBlock); - calBlockTbName(&pInfo->tbnameCalSup, pInfo->pRes); + calBlockTbName(pInfo, pInfo->pRes); return 0; } @@ -1960,9 +1979,13 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) { SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp); if (pBlock != NULL) { - calBlockTbName(&pInfo->tbnameCalSup, pBlock); - updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); + calBlockTbName(pInfo, pBlock); + if (pInfo->pUpdateInfo) { + TSKEY maxTs = updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); + pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs); + } qDebug("stream recover scan get block, rows %d", pBlock->info.rows); + printDataBlock(pBlock, "scan recover"); return pBlock; } pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE; @@ -2081,7 +2104,7 @@ FETCH_NEXT_BLOCK: pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA; checkUpdateData(pInfo, true, pSDB, false); // printDataBlock(pSDB, "stream scan update"); - calBlockTbName(&pInfo->tbnameCalSup, pSDB); + calBlockTbName(pInfo, pSDB); return pSDB; } blockDataCleanup(pInfo->pUpdateDataRes); @@ -2386,6 +2409,7 @@ static void destroyStreamScanOperatorInfo(void* param) { } cleanupExprSupp(&pStreamScan->tbnameCalSup); + taosHashCleanup(pStreamScan->pGroupIdTbNameMap); updateInfoDestroy(pStreamScan->pUpdateInfo); blockDataDestroy(pStreamScan->pRes); @@ -2443,6 +2467,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1) != 0) { goto _error; } + pInfo->pGroupIdTbNameMap = + taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK); } if (pTableScanNode->pTags != NULL) { diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c index f877bef44dda66077d8c737c208d57de8b510472..fe010786ebdcb12d9df1350fbe565c6fb1494e3f 100644 --- a/source/libs/function/src/builtins.c +++ b/source/libs/function/src/builtins.c @@ -917,6 +917,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err int32_t startIndex; if (numOfParams != 4) { snprintf(errMsg, msgLen, "%s", msg1); + cJSON_Delete(binDesc); return false; } @@ -928,17 +929,20 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { snprintf(errMsg, msgLen, "%s", msg3); + cJSON_Delete(binDesc); return false; } if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 snprintf(errMsg, msgLen, "%s", msg4); + cJSON_Delete(binDesc); return false; } if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { snprintf(errMsg, msgLen, "%s", msg5); + cJSON_Delete(binDesc); return false; } @@ -957,6 +961,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (width->valuedouble == 0) { snprintf(errMsg, msgLen, "%s", msg6); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { @@ -964,6 +969,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (isinf(intervals[startIndex])) { snprintf(errMsg, msgLen, "%s", msg5); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; @@ -973,11 +979,13 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (start->valuedouble == 0) { snprintf(errMsg, msgLen, "%s", msg7); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { snprintf(errMsg, msgLen, "%s", msg8); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { @@ -985,6 +993,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (isinf(intervals[startIndex])) { snprintf(errMsg, msgLen, "%s", msg5); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; @@ -992,6 +1001,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err } else { snprintf(errMsg, msgLen, "%s", msg3); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } @@ -1007,6 +1017,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err } else if (cJSON_IsArray(binDesc)) { /* user input bins */ if (binType != USER_INPUT_BIN) { snprintf(errMsg, msgLen, "%s", msg3); + cJSON_Delete(binDesc); return false; } numOfBins = cJSON_GetArraySize(binDesc); @@ -1015,6 +1026,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (bin == NULL) { snprintf(errMsg, msgLen, "%s", msg3); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } int i = 0; @@ -1023,11 +1035,13 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err if (!cJSON_IsNumber(bin)) { snprintf(errMsg, msgLen, "%s", msg3); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (i != 0 && intervals[i] <= intervals[i - 1]) { snprintf(errMsg, msgLen, "%s", msg3); taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } bin = bin->next; @@ -1035,6 +1049,7 @@ static bool validateHistogramBinDesc(char* binDescStr, int8_t binType, char* err } } else { snprintf(errMsg, msgLen, "%s", msg3); + cJSON_Delete(binDesc); return false; } @@ -1464,11 +1479,16 @@ static int32_t translateDerivative(SFunctionNode* pFunc, char* pErrBuf, int32_t uint8_t colType = ((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType.type; // param1 - SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 1); + SNode* pParamNode1 = nodesListGetNode(pFunc->pParameterList, 1); + SValueNode* pValue1 = (SValueNode*)pParamNode1; if (QUERY_NODE_VALUE != nodeType(pParamNode1)) { return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName); } + if (pValue1->datum.i <= 0) { + return invaildFuncParaValueErrMsg(pErrBuf, len, pFunc->functionName); + } + SValueNode* pValue = (SValueNode*)pParamNode1; pValue->notReserved = true; diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 2bd3555d1ab1849b75d93a33cffd1dcec4ddcc59..87e15370e4003189b08f41c2e57cdcc3b36e5925 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -1083,7 +1083,10 @@ int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { SAvgRes* pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); int32_t type = pAvgRes->type; - if (IS_SIGNED_NUMERIC_TYPE(type)) { + if (pAvgRes->count == 0) { + // [ASAN] runtime error: division by zero + GET_RES_INFO(pCtx)->numOfRes = 0; + } else if (IS_SIGNED_NUMERIC_TYPE(type)) { pAvgRes->result = pAvgRes->sum.isum / ((double)pAvgRes->count); } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) { pAvgRes->result = pAvgRes->sum.usum / ((double)pAvgRes->count); @@ -2924,6 +2927,8 @@ static void firstlastSaveTupleData(const SSDataBlock* pSrcBlock, int32_t rowInde if (!pInfo->hasResult) { pInfo->pos = saveTupleData(pCtx, rowIndex, pSrcBlock, NULL); + ASSERT(pCtx->subsidiaries.buf != NULL); + ASSERT(pCtx->subsidiaries.rowLen > 0); } else { updateTupleData(pCtx, rowIndex, pSrcBlock, &pInfo->pos); } @@ -4371,6 +4376,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t int32_t numOfParams = cJSON_GetArraySize(binDesc); int32_t startIndex; if (numOfParams != 4) { + cJSON_Delete(binDesc); return false; } @@ -4381,15 +4387,18 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t cJSON* infinity = cJSON_GetObjectItem(binDesc, "infinity"); if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { + cJSON_Delete(binDesc); return false; } if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 + cJSON_Delete(binDesc); return false; } if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { + cJSON_Delete(binDesc); return false; } @@ -4407,12 +4416,14 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t // linear bin process if (width->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble + i * width->valuedouble; if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; @@ -4421,22 +4432,26 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t // log bin process if (start->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; } } else { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } @@ -4451,6 +4466,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t } } else if (cJSON_IsArray(binDesc)) { /* user input bins */ if (binType != USER_INPUT_BIN) { + cJSON_Delete(binDesc); return false; } numOfBins = cJSON_GetArraySize(binDesc); @@ -4458,6 +4474,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t cJSON* bin = binDesc->child; if (bin == NULL) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } int i = 0; @@ -4465,16 +4482,19 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t intervals[i] = bin->valuedouble; if (!cJSON_IsNumber(bin)) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (i != 0 && intervals[i] <= intervals[i - 1]) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } bin = bin->next; i++; } } else { + cJSON_Delete(binDesc); return false; } @@ -4487,6 +4507,8 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t } taosMemoryFree(intervals); + cJSON_Delete(binDesc); + return true; } @@ -4500,18 +4522,23 @@ bool histogramFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultIn pInfo->totalCount = 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) { 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; if (normalized != 0 && normalized != 1) { return false; } if (!getHistogramBinDesc(pInfo, binDesc, binType, (bool)normalized)) { + taosMemoryFree(binDesc); return false; } + taosMemoryFree(binDesc); return true; } diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index e45beb7e13a24223468a95ca22d277e8485308fd..4caa266d8acbda5ce5f511267202ca3930916c9b 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -1191,6 +1191,8 @@ void qWorkerStopAllTasks(void *qWorkerMgmt) { if (QW_QUERY_RUNNING(ctx)) { qwKillTaskHandle(ctx); + } else if (!QW_EVENT_PROCESSED(ctx, QW_EVENT_DROP)) { + QW_SET_EVENT_RECEIVED(ctx, QW_EVENT_DROP); } QW_UNLOCK(QW_WRITE, &ctx->lock); diff --git a/source/libs/scalar/src/sclfunc.c b/source/libs/scalar/src/sclfunc.c index 22fa99d75d3229aba0f97d92c280b938f068875f..336bc6d533687d8ecaffa6e1c19c7f3c43357754 100644 --- a/source/libs/scalar/src/sclfunc.c +++ b/source/libs/scalar/src/sclfunc.c @@ -2620,6 +2620,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin int32_t numOfParams = cJSON_GetArraySize(binDesc); int32_t startIndex; if (numOfParams != 4) { + cJSON_Delete(binDesc); return false; } @@ -2630,15 +2631,18 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin cJSON *infinity = cJSON_GetObjectItem(binDesc, "infinity"); if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { + cJSON_Delete(binDesc); return false; } if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 + cJSON_Delete(binDesc); return false; } if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { + cJSON_Delete(binDesc); return false; } @@ -2656,12 +2660,14 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin // linear bin process if (width->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble + i * width->valuedouble; if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; @@ -2670,22 +2676,26 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin // log bin process if (start->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; } } else { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } @@ -2700,6 +2710,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin } } else if (cJSON_IsArray(binDesc)) { /* user input bins */ if (binType != USER_INPUT_BIN) { + cJSON_Delete(binDesc); return false; } numOfBins = cJSON_GetArraySize(binDesc); @@ -2707,6 +2718,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin cJSON *bin = binDesc->child; if (bin == NULL) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } int i = 0; @@ -2714,16 +2726,19 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin intervals[i] = bin->valuedouble; if (!cJSON_IsNumber(bin)) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (i != 0 && intervals[i] <= intervals[i - 1]) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } bin = bin->next; i++; } } else { + cJSON_Delete(binDesc); return false; } @@ -2735,8 +2750,9 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin (*bins)[i].count = 0; } - cJSON_Delete(binDesc); taosMemoryFree(intervals); + cJSON_Delete(binDesc); + return true; } @@ -2748,14 +2764,19 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP int32_t numOfBins = 0; int32_t totalCount = 0; - int8_t binType = getHistogramBinType(varDataVal(pInput[1].columnData->pData)); - char *binDesc = varDataVal(pInput[2].columnData->pData); + char *binTypeStr = strndup(varDataVal(pInput[1].columnData->pData), varDataLen(pInput[1].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); int32_t type = GET_PARAM_TYPE(pInput); if (!getHistogramBinDesc(&bins, &numOfBins, binDesc, binType, (bool)normalized)) { + taosMemoryFree(binDesc); return TSDB_CODE_FAILED; } + taosMemoryFree(binDesc); for (int32_t i = 0; i < pInput->numOfRows; ++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) { int32_t len; char buf[512] = {0}; diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index e6d5859163f56150d4ede3a3903daaf79a3be4f5..79549675a356ddc716dc6a5a87e26bb3bdcd005c 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -51,6 +51,7 @@ void streamSchedByTimer(void* param, void* tmrId) { SStreamTask* pTask = (void*)param; if (atomic_load_8(&pTask->taskStatus) == TASK_STATUS__DROPPING) { + streamMetaReleaseTask(NULL, pTask); return; } @@ -80,6 +81,8 @@ void streamSchedByTimer(void* param, void* tmrId) { int32_t streamSetupTrigger(SStreamTask* pTask) { 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->triggerStatus = TASK_TRIGGER_STATUS__INACTIVE; } diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index e7f2b60704b7c6d17af8390b6d724fed77dbbd44..009f7eec9a2c28695adb28c34c3632912b7ed1ee 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -91,6 +91,7 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) { void* exec = pTask->exec.executor; qSetStreamOpOpen(exec); + bool finished = false; while (1) { SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock)); @@ -106,7 +107,10 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) { if (qExecTask(exec, &output, &ts) < 0) { ASSERT(0); } - if (output == NULL) break; + if (output == NULL) { + finished = true; + break; + } SSDataBlock block = {0}; assignOneDataBlock(&block, output); @@ -133,6 +137,7 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) { if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) { streamDispatch(pTask); } + if (finished) break; } return 0; } diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index 5ec8828c05203b4df7d3c3cc2750593729cb9c5f..a864814a7472ac347f39263ca1f32ec5a29d5a51 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -80,7 +80,12 @@ void streamMetaClose(SStreamMeta* pMeta) { pIter = taosHashIterate(pMeta->pTasks, pIter); if (pIter == NULL) break; SStreamTask* pTask = *(SStreamTask**)pIter; + if (pTask->timer) { + taosTmrStop(pTask->timer); + pTask->timer = NULL; + } tFreeSStreamTask(pTask); + /*streamMetaReleaseTask(pMeta, pTask);*/ } taosHashCleanup(pMeta->pTasks); taosMemoryFree(pMeta->path); @@ -202,6 +207,10 @@ void streamMetaRemoveTask1(SStreamMeta* pMeta, int32_t taskId) { if (ppTask) { SStreamTask* pTask = *ppTask; taosHashRemove(pMeta->pTasks, &taskId, sizeof(int32_t)); + /*if (pTask->timer) { + * taosTmrStop(pTask->timer);*/ + /*pTask->timer = NULL;*/ + /*}*/ atomic_store_8(&pTask->taskStatus, TASK_STATUS__DROPPING); taosWLockLatch(&pMeta->lock); diff --git a/source/libs/stream/src/streamUpdate.c b/source/libs/stream/src/streamUpdate.c index 199892c24196b5178981a5b400cf62a4d5d02e88..15526cd8bbfd36c6dfaa743979c91c550e413868 100644 --- a/source/libs/stream/src/streamUpdate.c +++ b/source/libs/stream/src/streamUpdate.c @@ -163,9 +163,9 @@ bool updateInfoIsTableInserted(SUpdateInfo *pInfo, int64_t tbUid) { return false; } -void updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t primaryTsCol) { - if (pBlock == NULL || pBlock->info.rows == 0) return; - TSKEY maxTs = -1; +TSKEY updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t primaryTsCol) { + if (pBlock == NULL || pBlock->info.rows == 0) return INT64_MIN; + TSKEY maxTs = INT64_MIN; int64_t tbUid = pBlock->info.uid; SColumnInfoData *pColDataInfo = taosArrayGet(pBlock->pDataBlock, primaryTsCol); @@ -186,6 +186,7 @@ void updateInfoFillBlockData(SUpdateInfo *pInfo, SSDataBlock *pBlock, int32_t pr if (pMaxTs == NULL || *pMaxTs > maxTs) { taosHashPut(pInfo->pMap, &tbUid, sizeof(int64_t), &maxTs, sizeof(TSKEY)); } + return maxTs; } bool updateInfoIsUpdated(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts) { diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index e4d4b28b6b86b6a2c1a05197073cf6e081bdfe14..362618fece3734197280dfdda061fab310407da0 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -73,11 +73,12 @@ typedef struct SSyncTimer { SSyncHbTimerData hbData; } SSyncTimer; -typedef struct SElectTimer { +typedef struct SElectTimerParam { uint64_t logicClock; SSyncNode* pSyncNode; + int64_t executeTime; void* pData; -} SElectTimer; +} SElectTimerParam; typedef struct SPeerState { SyncIndex lastSendIndex; @@ -153,6 +154,7 @@ typedef struct SSyncNode { uint64_t electTimerLogicClock; TAOS_TMR_CALLBACK FpElectTimerCB; // Timer Fp uint64_t electTimerCounter; + SElectTimerParam electTimerParam; // heartbeat timer tmr_h pHeartbeatTimer; diff --git a/source/libs/sync/src/syncAppendEntriesReply.c b/source/libs/sync/src/syncAppendEntriesReply.c index 875204c0a9799386c069872b6d9a9aace85fb230..13ea250155cca006170725af991ba5da69d65d67 100644 --- a/source/libs/sync/src/syncAppendEntriesReply.c +++ b/source/libs/sync/src/syncAppendEntriesReply.c @@ -83,10 +83,14 @@ int32_t syncNodeOnAppendEntriesReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) { ASSERT(pState != NULL); 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); } } syncLogRecvAppendEntriesReply(ths, pMsg, "process"); return 0; -} \ No newline at end of file +} diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 514160235f48dd838482d15400a0c44458c3c56c..0370b3c5b1966fd2f0106e4596e0647b11d924a2 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -429,7 +429,7 @@ bool syncIsReadyForRead(int64_t rid) { int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) { 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; return -1; } @@ -445,7 +445,7 @@ int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) { int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) { 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; return -1; } @@ -1104,8 +1104,16 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t ret = 0; if (syncIsInit()) { 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, &pSyncNode->pElectTimer); + } else { 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) { - SSyncNode* pNode = param; - if (!syncIsInit()) return; + + SSyncNode* pNode = (SSyncNode*)param; + if (pNode == NULL) return; if (pNode->syncEqMsg == NULL) return; + int64_t tsNow = taosGetTimestampMs(); + if (tsNow < pNode->electTimerParam.executeTime) return; + SRpcMsg rpcMsg = {0}; 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) { sError("failed to build elect msg"); + return; } 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); if (code != 0) { sError("failed to sync enqueue elect msg since %s", terrstr()); rpcFreeCont(rpcMsg.pCont); + + return; } } @@ -2059,6 +2075,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) { SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont; pSyncMsg->cmd = SYNC_LOCAL_CMD_FOLLOWER_CMT; pSyncMsg->fcIndex = pMsg->commitIndex; + SyncIndex fcIndex = pSyncMsg->fcIndex; if (ths->syncEqMsg != NULL && ths->msgcb != NULL) { 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); rpcFreeCont(rpcMsgLocalCmd.pCont); } 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); } } } diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index 9bc0a07d48fd95b33768c859320eaad8816a00d0..222b7c4e1ef97a3c82fcb419597875ef4f1dab0d 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -192,7 +192,9 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { // pMsg->privateTerm = pSender->privateTerm; - memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); + if (pSender->pCurrentBlock != NULL) { + memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen); + } // send msg syncNodeSendMsgById(&pMsg->destId, pSender->pSyncNode, &rpcMsg); diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index fb1b07b0b62cab4ef72221827d4507a91a569eee..b50336cd63e5526e99155bb858df11666783723c 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -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, ...) { - 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; // 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, const char* format, ...) { 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}; 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, const char* format, ...) { 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}; 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); sNTrace(pSyncNode, "send sync-request-vote-reply to %s:%d {term:%" PRId64 ", grant:%d}, %s", host, port, pMsg->term, pMsg->voteGranted, s); -} \ No newline at end of file +} diff --git a/source/libs/wal/src/walMgmt.c b/source/libs/wal/src/walMgmt.c index b78d5ca6c1c0f4573e473d78ed7d784dc2297d6b..5d0f020b02cfc62a23ffbf3de2dc656e7d150a9b 100644 --- a/source/libs/wal/src/walMgmt.c +++ b/source/libs/wal/src/walMgmt.c @@ -251,7 +251,7 @@ static void walFsyncAll() { int32_t code = taosFsyncFile(pWal->pLogFile); if (code != 0) { wError("vgId:%d, file:%" PRId64 ".log, failed to fsync since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal), - strerror(code)); + strerror(errno)); } } pWal = taosIterateRef(tsWal.refSetId, pWal->refId); diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 5067f7a8266806481255447bf9aba935ca6fa732..6cd127fed4eb354f79c7f444cde1cfd95caaa1ee 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -102,6 +102,7 @@ int32_t walCommit(SWal *pWal, int64_t ver) { int32_t walRollback(SWal *pWal, int64_t ver) { taosThreadMutexLock(&pWal->mutex); + wInfo("vgId:%d, wal rollback for version %" PRId64, pWal->cfg.vgId, ver); int64_t code; char fnameStr[WAL_FILE_LEN]; 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); for (int i = pWal->writeCur + 1; i < fileSetSize; i++) { walBuildLogName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr); + wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr); taosRemoveFile(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); } // pop from fileInfoSet @@ -157,6 +160,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) { walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr); 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) { // TODO terrno = TAOS_SYSTEM_ERROR(errno); @@ -321,12 +325,12 @@ int32_t walEndSnapshot(SWal *pWal) { SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE); if (pInfo) { if (ver >= pInfo->lastVer) { - pInfo++; + pInfo--; } 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 { - wDebug("vgId:%d, no remove", pWal->cfg.vgId); + wDebug("vgId:%d, wal no remove", pWal->cfg.vgId); } // iterate files, until the searched result 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++) { pInfo = taosArrayGet(pWal->fileInfoSet, i); 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) { goto UPDATE_META; } 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) { ASSERT(0); } diff --git a/source/util/src/terror.c b/source/util/src/terror.c index b4064326169aa5ee8768e0f0823ba2c7eb0c41d4..43602a607a771d11ba15a6d46cafe830917ddeb6 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -646,7 +646,12 @@ const char* tstrerror(int32_t err) { // this is a system errno if ((err & 0x00ff0000) == 0x00ff0000) { - return strerror(err & 0x0000ffff); + int32_t code = err & 0x0000ffff; + if (code >= 0 && code < 36) { + return strerror(code); + } else { + return "unknown err"; + } } int32_t s = 0; diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index b9d5e6c03f9b51dbe8531b9c8fa4beec9d5bf4a1..7115e122526a852909e46edfdaa1b703ecac7171 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -11,7 +11,7 @@ ,,y,script,./test.sh -f tsim/user/privilege_db.sim ,,y,script,./test.sh -f tsim/user/privilege_sysinfo.sim ,,y,script,./test.sh -f tsim/db/alter_option.sim -,,,script,./test.sh -f tsim/db/alter_replica_13.sim +,,y,script,./test.sh -f tsim/db/alter_replica_13.sim ,,y,script,./test.sh -f tsim/db/alter_replica_31.sim ,,y,script,./test.sh -f tsim/db/basic1.sim ,,y,script,./test.sh -f tsim/db/basic2.sim @@ -115,9 +115,9 @@ ,,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/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 -,,,script,./test.sh -f tsim/parser/groupby.sim +,,y,script,./test.sh -f tsim/parser/groupby.sim ,,y,script,./test.sh -f tsim/parser/having_child.sim ,,y,script,./test.sh -f tsim/parser/having.sim ,,y,script,./test.sh -f tsim/parser/import_commit1.sim @@ -140,7 +140,7 @@ ,,,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/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/precision_ns.sim ,,y,script,./test.sh -f tsim/parser/projection_limit_offset.sim @@ -220,7 +220,7 @@ ,,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/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/state0.sim ,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim diff --git a/tests/script/test.sh b/tests/script/test.sh index 1e3319dd0ad634943d0e4f2d89766a4d6852e03a..e1db1c14de3ab8bd72b86d649bbc25e83bde139f 100755 --- a/tests/script/test.sh +++ b/tests/script/test.sh @@ -7,6 +7,7 @@ ################################################## set +e +#set -x FILE_NAME= RELEASE=0 @@ -137,8 +138,16 @@ if [ -n "$FILE_NAME" ]; then $PROGRAM -c $CFG_DIR -f $FILE_NAME -v else echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f $FILE_NAME - $PROGRAM -c $CFG_DIR -f $FILE_NAME 2> $ASAN_DIR/tsim.asan - $CODE_DIR/sh/checkAsan.sh + echo "AsanDir:" $ASAN_DIR/tsim.asan + 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 else echo "ExcuteCmd:" $PROGRAM -c $CFG_DIR -f basicSuite.sim diff --git a/tests/script/tsim/compute/diff2.sim b/tests/script/tsim/compute/diff2.sim index 1cc2a87839f84f2b742fbb0667c373ad1923ea15..5848e05b1e8d91016605e0514141269f32c8c4d6 100644 --- a/tests/script/tsim/compute/diff2.sim +++ b/tests/script/tsim/compute/diff2.sim @@ -83,7 +83,7 @@ sql select diff(c1), diff(c2) from $tb sql select 2+diff(c1) 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 select diff(c1) from $mt +#sql select diff(c1) from $mt sql_error select diff(diff(c1)) from $tb sql_error select diff(c1) from m_di_tb1 where c2 like '2%' diff --git a/tests/script/tsim/parser/function.sim b/tests/script/tsim/parser/function.sim index ec9cdf2666617c52192685a445d0d7334d4a6387..7f69aa2d02adeaa2cd64eafbefaa2d2df3a79184 100644 --- a/tests/script/tsim/parser/function.sim +++ b/tests/script/tsim/parser/function.sim @@ -821,10 +821,10 @@ sql insert into tm0 values('2015-08-18T00:18:00Z', 2.126) ('2015-08-18T00:24:00Z sql_error select derivative(ts) from tm0; sql_error select derivative(k) from tm0; -sql select derivative(k, 0, 0) from tm0; +sql_error select derivative(k, 0, 0) from tm0; sql_error select derivative(k, 1, 911) from tm0; sql_error select derivative(kx, 1s, 1) from tm0; -sql select derivative(k, -20s, 1) from tm0; +sql_error select derivative(k, -20s, 1) from tm0; sql select derivative(k, 20a, 0) from tm0; sql select derivative(k, 200a, 0) from tm0; sql select derivative(k, 999a, 0) from tm0; @@ -932,7 +932,7 @@ sql insert into t0 values('2020-1-1 1:4:10', 10); sql insert into t1 values('2020-1-1 1:1:2', 2); print ===========================>td-4739 -sql select diff(val) from (select derivative(k, 1s, 0) val from t1); +sql select diff(val) from (select ts, derivative(k, 1s, 0) val from t1); if $rows != 0 then return -1 endi @@ -1077,4 +1077,4 @@ endi if $data11 != NULL then print ======data11=$data11 return -1 -endi \ No newline at end of file +endi diff --git a/tests/script/tsim/parser/nestquery.sim b/tests/script/tsim/parser/nestquery.sim index dd7a0c6c81a14430d18beabe70efee6118fe33d9..494c3de99f473f1b72eab3e465e1d506d5470152 100644 --- a/tests/script/tsim/parser/nestquery.sim +++ b/tests/script/tsim/parser/nestquery.sim @@ -338,13 +338,13 @@ if $data03 != @20-09-15 00:00:00.000@ then return -1 endi -sql select diff(val) from (select c1 val from nest_tb0); -if $rows != 9999 then - return -1 -endi -if $data00 != 1 then - return -1 -endi +#sql select diff(val) from (select c1 val from nest_tb0); +#if $rows != 9999 then +# return -1 +#endi +#if $data00 != 1 then +# return -1 +#endi sql_error select last_row(*) from (select * from nest_tb0) having c1 > 0 diff --git a/tests/script/tsim/valgrind/checkError6.sim b/tests/script/tsim/valgrind/checkError6.sim index 89b6cadef88eb5a7d363b5fd6c784b7606095e66..ef8fe930b913eb196874bbd11a5055b9cc832761 100644 --- a/tests/script/tsim/valgrind/checkError6.sim +++ b/tests/script/tsim/valgrind/checkError6.sim @@ -92,7 +92,7 @@ sql select avg(tbcol) as b from stb where ts <= 1601481840000 partition by tgcol sql show table distributed 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 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 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) @@ -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 show table distributed stb 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 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)