diff --git a/docs/en/12-taos-sql/22-meta.md b/docs/en/12-taos-sql/22-meta.md index 2d7c517dca..37304633e7 100644 --- a/docs/en/12-taos-sql/22-meta.md +++ b/docs/en/12-taos-sql/22-meta.md @@ -297,3 +297,13 @@ Provides dnode configuration information. | 7 | target_table | BINARY(192) | Target table | | 8 | watermark | BIGINT | Watermark (see stream processing documentation). It should be noted that `watermark` is a TDengine keyword and needs to be escaped with ` when used as a column name. | | 9 | trigger | INT | Method of triggering the result push (see stream processing documentation). It should be noted that `trigger` is a TDengine keyword and needs to be escaped with ` when used as a column name. | + +## INS_USER_PRIVILEGES + +| # | **Column** | **Data Type** | **Description** |** | +| --- | :----------: | ------------ | -------------------------------------------| +| 1 | user_name | VARCHAR(24) | Username | +| 2 | privilege | VARCHAR(10) | Privilege description | +| 3 | db_name | VARCHAR(65) | Database name | +| 4 | table_name | VARCHAR(193) | Table name | +| 5 | condition | VARCHAR(49152) | The privilege filter for child tables | diff --git a/docs/en/14-reference/12-config/index.md b/docs/en/14-reference/12-config/index.md index a080ef75b4..c0512aef7b 100755 --- a/docs/en/14-reference/12-config/index.md +++ b/docs/en/14-reference/12-config/index.md @@ -670,6 +670,15 @@ The charset that takes effect is UTF-8. | Value Range | 0: not consistent; 1: consistent. | | Default | 0 | +### smlTsDefaultName + +| Attribute | Description | +| -------- | -------------------------------------------------------- | +| Applicable | Client only | +| Meaning | The name of the time column for schemaless automatic table creation is set through this configuration | +| Type | String | +| Default Value | _ts | + ## Compress Parameters ### compressMsgSize diff --git a/docs/en/14-reference/13-schemaless/13-schemaless.md b/docs/en/14-reference/13-schemaless/13-schemaless.md index 3ae9098a73..54be18eea3 100644 --- a/docs/en/14-reference/13-schemaless/13-schemaless.md +++ b/docs/en/14-reference/13-schemaless/13-schemaless.md @@ -34,7 +34,27 @@ In the schemaless writing data line protocol, each data item in the field_set ne - If there are English double quotes on both sides, it indicates the BINARY(32) type. For example, `"abc"`. - If there are double quotes on both sides and an L prefix, it means NCHAR(32) type. For example, `L"error message"`. -- Spaces, equal signs (=), commas (,), and double quotes (") need to be escaped with a backslash (\\) in front. (All refer to the ASCII character) +- Spaces, equals sign (=), comma (,), double quote ("), and backslash (\\) need to be escaped with a backslash (\\) in front. (All refer to the ASCII character). The rules are as follows: + +| **Serial number** | **Element** | **Escape characters** | +| -------- | ----------- | ----------------------------- | +| 1 | Measurement | Comma, Space | +| 2 | Tag key | Comma, Equals Sign, Space | +| 3 | Tag value | Comma, Equals Sign, Space | +| 4 | Field key | Comma, Equals Sign, Space | +| 5 | Field value | Double quote, Backslash | + +With two contiguous backslashes, the first is interpreted as an escape character. Examples of backslash escape rules are as follows: + +| **Serial number** | **Backslashes** | **Interpreted as** | +| -------- | ----------- | ----------------------------- | +| 1 | \ | \ | +| 2 | \\\\ | \ | +| 3 | \\\\\\ | \\\\ | +| 4 | \\\\\\\\ | \\\\ | +| 5 | \\\\\\\\\\ | \\\\\\ | +| 6 | \\\\\\\\\\\\ | \\\\\\ | + - Numeric types will be distinguished from data types by the suffix. | **Serial number** | **Postfix** | **Mapping type** | **Size (bytes)** | @@ -88,6 +108,8 @@ You can configure smlChildTableName in taos.cfg to specify table names, for exam 8. It is assumed that the order of field_set in a supertable is consistent, meaning that the first record contains all fields and subsequent records store fields in the same order. If the order is not consistent, set smlDataFormat in taos.cfg to false. Otherwise, data will be written out of order and a database error will occur. Note: TDengine 3.0.3.0 and later automatically detect whether order is consistent. This parameter is no longer used. +9. Due to the fact that SQL table names do not support period (.), schemaless has also processed period (.). If there is a period (.) in the table name automatically created by schemaless, it will be automatically replaced with an underscore (\_). If you manually specify a sub table name, if there is a dot (.) in the sub table name, it will also be converted to an underscore (\_) +10. Taos.cfg adds the configuration of smlTsDefaultName (with a string value), which only works on the client side. After configuration, the time column name of the schemaless automatic table creation can be set through this configuration. If not configured, defaults to _ts. :::tip All processing logic of schemaless will still follow TDengine's underlying restrictions on data structures, such as the total length of each row of data cannot exceed 48 KB(64 KB since version 3.0.5.0) and the total length of a tag value cannot exceed 16 KB. See [TDengine SQL Boundary Limits](/taos-sql/limit) for specific constraints in this area. diff --git a/docs/zh/12-taos-sql/22-meta.md b/docs/zh/12-taos-sql/22-meta.md index a5396d8607..35794ec269 100644 --- a/docs/zh/12-taos-sql/22-meta.md +++ b/docs/zh/12-taos-sql/22-meta.md @@ -298,3 +298,13 @@ TDengine 内置了一个名为 `INFORMATION_SCHEMA` 的数据库，提供对数 | 7 | target_table | BINARY(192) | 流计算写入的目标表 | | 8 | watermark | BIGINT | watermark，详见 SQL 手册流式计算。需要注意，`watermark` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。 | | 9 | trigger | INT | 计算结果推送模式，详见 SQL 手册流式计算。需要注意，`trigger` 为 TDengine 关键字，作为列名使用时需要使用 ` 进行转义。 | + +## INS_USER_PRIVILEGES + +| # | **列名** | **数据类型** | **说明** | +| --- | :----------: | ------------ | -------------------------------------------------------------------------------------------------------------------- | +| 1 | user_name | VARCHAR(24) | 用户名 +| 2 | privilege | VARCHAR(10) | 权限描述 +| 3 | db_name | VARCHAR(65) | 数据库名称 +| 4 | table_name | VARCHAR(193) | 表名称 +| 5 | condition | VARCHAR(49152) | 子表权限过滤条件 diff --git a/docs/zh/14-reference/12-config/index.md b/docs/zh/14-reference/12-config/index.md index ba5eb63859..2f5f0fc3e8 100755 --- a/docs/zh/14-reference/12-config/index.md +++ b/docs/zh/14-reference/12-config/index.md @@ -685,7 +685,16 @@ charset 的有效值是 UTF-8。 | 适用范围 | 仅客户端适用 | | 含义 | schemaless 列数据是否顺序一致，从3.0.3.0开始，该配置废弃 | | 值域 | 0：不一致；1: 一致 | -| 缺省值 | 0 | +| 缺省值 | 0 + +### smlTsDefaultName + +| 属性 | 说明 | +| -------- | -------------------------------------------------------- | +| 适用范围 | 仅客户端适用 | +| 含义 | schemaless自动建表的时间列名字通过该配置设置 | +| 类型 | 字符串 | +| 缺省值 | _ts | ## 其他 diff --git a/docs/zh/14-reference/13-schemaless/13-schemaless.md b/docs/zh/14-reference/13-schemaless/13-schemaless.md index 6c2007938b..9f5bae081c 100644 --- a/docs/zh/14-reference/13-schemaless/13-schemaless.md +++ b/docs/zh/14-reference/13-schemaless/13-schemaless.md @@ -35,12 +35,32 @@ tag_set 中的所有的数据自动转化为 nchar 数据类型，并不需要 - 如果两边有英文双引号，表示 BINARY(32) 类型。例如 `"abc"`。 - 如果两边有英文双引号而且带有 L 前缀，表示 NCHAR(32) 类型。例如 `L"报错信息"`。 -- 对空格、等号（=）、逗号（,）、双引号（"），前面需要使用反斜杠（\）进行转义。（都指的是英文半角符号） +- 对空格、等号（=）、逗号（,）、双引号（"）、反斜杠（\），前面需要使用反斜杠（\）进行转义。（都指的是英文半角符号）。具体转义规则如下： + +| **序号** | **域** | **需转义字符** | +| -------- | ----------- | ----------------------------- | +| 1 | 超级表名 | 逗号，空格 | +| 2 | 标签名 | 逗号，等号，空格 | +| 3 | 标签值 | 逗号，等号，空格 | +| 4 | 列名 | 逗号，等号，空格 | +| 5 | 列值 | 双引号，反斜杠 | + +两个连续的反斜杠，第一个作为转义符，只有一个反斜杠则无需转义. 反斜杠转义规则举例如下: + +| **序号** | **反斜杠** | **转义为** | +| -------- | ----------- | ----------------------------- | +| 1 | \ | \ | +| 2 | \\\\ | \ | +| 3 | \\\\\\ | \\\\ | +| 4 | \\\\\\\\ | \\\\ | +| 5 | \\\\\\\\\\ | \\\\\\ | +| 6 | \\\\\\\\\\\\ | \\\\\\ | + - 数值类型将通过后缀来区分数据类型： -| **序号** | **后缀** | **映射类型** | **大小(字节)** | +| **序号** | **后缀** | **映射类型** | **大小(字节)** | | -------- | ----------- | ----------------------------- | -------------- | -| 1 | 无或 f64 | double | 8 | +| 1 | 无或 f64 | double | 8 | | 2 | f32 | float | 4 | | 3 | i8/u8 | TinyInt/UTinyInt | 1 | | 4 | i16/u16 | SmallInt/USmallInt | 2 | @@ -84,7 +104,9 @@ st,t1=3,t2=4,t3=t3 c1=3i64,c3="passit",c2=false,c4=4f64 1626006833639000000 6. 对 BINARY 或 NCHAR 列，如果数据行中所提供值的长度超出了列类型的限制，自动增加该列允许存储的字符长度上限（只增不减），以保证数据的完整保存。 7. 整个处理过程中遇到的错误会中断写入过程，并返回错误代码。 8. 为了提高写入的效率，默认假设同一个超级表中 field_set 的顺序是一样的（第一条数据包含所有的 field，后面的数据按照这个顺序），如果顺序不一样，需要配置参数 smlDataFormat 为 false，否则，数据写入按照相同顺序写入，库中数据会异常，从3.0.3.0开始，自动检测顺序是否一致，该配置废弃。 - +9. 由于sql建表表名不支持点号（.），所以schemaless也对点号（.）做了处理，如果schemaless自动建表的表名如果有点号（.），会自动替换为下划线（\_）。如果手动指定子表名的话，子表名里有点号（.），同样转化为下划线（\_）。 +10. taos.cfg 增加 smlTsDefaultName 配置（值为字符串），只在client端起作用，配置后，schemaless自动建表的时间列名字可以通过该配置设置。不配置的话，默认为 _ts + :::tip 无模式所有的处理逻辑，仍会遵循 TDengine 对数据结构的底层限制，例如每行数据的总长度不能超过 48KB（从 3.0.5.0 版本开始为 64KB），标签值的总长度不超过16KB。这方面的具体限制约束请参见 [TDengine SQL 边界限制](/taos-sql/limit) diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 4f2ed2b065..8ea0a857e8 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -169,6 +169,8 @@ extern char tsUdfdLdLibPath[]; // schemaless extern char tsSmlChildTableName[]; extern char tsSmlTagName[]; +extern bool tsSmlDot2Underline; +extern char tsSmlTsDefaultName[]; // extern bool tsSmlDataFormat; // extern int32_t tsSmlBatchSize; diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h index 2fbd7851e8..773f373a2d 100644 --- a/include/libs/executor/storageapi.h +++ b/include/libs/executor/storageapi.h @@ -368,6 +368,8 @@ typedef struct SStateStore { bool (*updateInfoIsUpdated)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts); bool (*updateInfoIsTableInserted)(SUpdateInfo* pInfo, int64_t tbUid); void (*updateInfoDestroy)(SUpdateInfo* pInfo); + void (*windowSBfDelete)(SUpdateInfo *pInfo, uint64_t count); + void (*windowSBfAdd)(SUpdateInfo *pInfo, uint64_t count); SUpdateInfo* (*updateInfoInitP)(SInterval* pInterval, int64_t watermark); void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo); diff --git a/include/libs/stream/tstreamUpdate.h b/include/libs/stream/tstreamUpdate.h index bd5a3be8de..7bb1d027c9 100644 --- a/include/libs/stream/tstreamUpdate.h +++ b/include/libs/stream/tstreamUpdate.h @@ -53,6 +53,8 @@ void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo); void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo); int32_t updateInfoSerialize(void *buf, int32_t bufLen, const SUpdateInfo *pInfo); int32_t updateInfoDeserialize(void *buf, int32_t bufLen, SUpdateInfo *pInfo); +void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count); +void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count); #ifdef __cplusplus } diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 2b8c72b829..b43985074c 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -780,6 +780,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_TMQ_INVALID_TOPIC TAOS_DEF_ERROR_CODE(0, 0x4009) #define TSDB_CODE_TMQ_NEED_INITIALIZED TAOS_DEF_ERROR_CODE(0, 0x4010) #define TSDB_CODE_TMQ_NO_COMMITTED TAOS_DEF_ERROR_CODE(0, 0x4011) +#define TSDB_CODE_TMQ_SAME_COMMITTED_VALUE TAOS_DEF_ERROR_CODE(0, 0x4012) // stream #define TSDB_CODE_STREAM_TASK_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x4100) diff --git a/include/util/tdef.h b/include/util/tdef.h index 3ee35b3b6b..e4af88bf10 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -200,7 +200,7 @@ typedef enum ELogicConditionType { #define TSDB_STREAM_NAME_LEN 193 // it is a null-terminated string #define TSDB_DB_NAME_LEN 65 #define TSDB_DB_FNAME_LEN (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN) -#define TSDB_PRIVILEDGE_CONDITION_LEN 200 +#define TSDB_PRIVILEDGE_CONDITION_LEN 48*1024 #define TSDB_FUNC_NAME_LEN 65 #define TSDB_FUNC_COMMENT_LEN 1024 * 1024 @@ -369,8 +369,13 @@ typedef enum ELogicConditionType { #define TSDB_DB_SCHEMALESS_OFF 0 #define TSDB_DEFAULT_DB_SCHEMALESS TSDB_DB_SCHEMALESS_OFF #define TSDB_MIN_STT_TRIGGER 1 -#define TSDB_MAX_STT_TRIGGER 16 -#define TSDB_DEFAULT_SST_TRIGGER 2 +#ifdef TD_ENTERPRISE +#define TSDB_MAX_STT_TRIGGER 16 +#define TSDB_DEFAULT_SST_TRIGGER 2 +#else +#define TSDB_MAX_STT_TRIGGER 1 +#define TSDB_DEFAULT_SST_TRIGGER 1 +#endif #define TSDB_MIN_HASH_PREFIX (2 - TSDB_TABLE_NAME_LEN) #define TSDB_MAX_HASH_PREFIX (TSDB_TABLE_NAME_LEN - 2) #define TSDB_DEFAULT_HASH_PREFIX 0 diff --git a/source/client/inc/clientSml.h b/source/client/inc/clientSml.h index c9eb951014..040064560c 100644 --- a/source/client/inc/clientSml.h +++ b/source/client/inc/clientSml.h @@ -64,8 +64,8 @@ extern "C" { #define IS_INVALID_COL_LEN(len) ((len) <= 0 || (len) >= TSDB_COL_NAME_LEN) #define IS_INVALID_TABLE_LEN(len) ((len) <= 0 || (len) >= TSDB_TABLE_NAME_LEN) -#define TS "_ts" -#define TS_LEN 3 +//#define TS "_ts" +//#define TS_LEN 3 #define VALUE "_value" #define VALUE_LEN 6 @@ -258,6 +258,7 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements); int32_t smlParseJSON(SSmlHandle *info, char *payload); +void smlStrReplace(char* src, int32_t len); #ifdef __cplusplus } #endif diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index 13dc019feb..ffff3df5d0 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -104,7 +104,7 @@ static int32_t smlCheckAuth(SSmlHandle *info, SRequestConnInfo* conn, const cha SUserAuthRes authRes = {0}; code = catalogChkAuth(info->pCatalog, conn, &pAuth, &authRes); - + nodesDestroyNode(authRes.pCond); return (code == TSDB_CODE_SUCCESS) ? (authRes.pass ? TSDB_CODE_SUCCESS : TSDB_CODE_PAR_PERMISSION_DENIED) : code; @@ -114,6 +114,15 @@ inline bool smlDoubleToInt64OverFlow(double num) { return false; } +void smlStrReplace(char* src, int32_t len){ + if (!tsSmlDot2Underline) return; + for(int i = 0; i < len; i++){ + if(src[i] == '.'){ + src[i] = '_'; + } + } +} + int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2) { if (pBuf->buf) { memset(pBuf->buf, 0, pBuf->len); @@ -193,6 +202,9 @@ static int32_t smlParseTableName(SArray *tags, char *childTableName) { if (childTableNameLen == tag->keyLen && strncmp(tag->key, tsSmlChildTableName, tag->keyLen) == 0) { memset(childTableName, 0, TSDB_TABLE_NAME_LEN); strncpy(childTableName, tag->value, (tag->length < TSDB_TABLE_NAME_LEN ? tag->length : TSDB_TABLE_NAME_LEN)); + if(tsSmlDot2Underline){ + smlStrReplace(childTableName, strlen(childTableName)); + } taosArrayRemove(tags, i); break; } @@ -838,6 +850,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) { char *measure = taosMemoryMalloc(superTableLen); memcpy(measure, superTable, superTableLen); PROCESS_SLASH_IN_MEASUREMENT(measure, superTableLen); + smlStrReplace(measure, superTableLen); memset(pName.tname, 0, TSDB_TABLE_NAME_LEN); memcpy(pName.tname, measure, superTableLen); taosMemoryFree(measure); @@ -1051,7 +1064,8 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) { taosMemoryFreeClear(sTableData->tableMeta); sTableData->tableMeta = pTableMeta; uDebug("SML:0x%" PRIx64 "modify schema uid:%" PRIu64 ", sversion:%d, tversion:%d", info->id, pTableMeta->uid, - pTableMeta->sversion, pTableMeta->tversion) tmp = (SSmlSTableMeta **)taosHashIterate(info->superTables, tmp); + pTableMeta->sversion, pTableMeta->tversion); + tmp = (SSmlSTableMeta **)taosHashIterate(info->superTables, tmp); } uDebug("SML:0x%" PRIx64 " smlModifyDBSchemas end success, format:%d, needModifySchema:%d", info->id, info->dataFormat, info->needModifySchema); @@ -1394,7 +1408,14 @@ static int32_t smlInsertData(SSmlHandle *info) { SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL); while (oneTable) { SSmlTableInfo *tableData = *oneTable; - tstrncpy(pName.tname, tableData->sTableName, tableData->sTableNameLen + 1); + + int measureLen = tableData->sTableNameLen; + char *measure = (char *)taosMemoryMalloc(tableData->sTableNameLen); + memcpy(measure, tableData->sTableName, tableData->sTableNameLen); + PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen); + smlStrReplace(measure, measureLen); + memset(pName.tname, 0, TSDB_TABLE_NAME_LEN); + memcpy(pName.tname, measure, measureLen); if (info->pRequest->tableList == NULL) { info->pRequest->tableList = taosArrayInit(1, sizeof(SName)); @@ -1411,6 +1432,7 @@ static int32_t smlInsertData(SSmlHandle *info) { code = smlCheckAuth(info, &conn, pName.tname, AUTH_TYPE_WRITE); if(code != TSDB_CODE_SUCCESS){ + taosMemoryFree(measure); return code; } @@ -1418,6 +1440,7 @@ static int32_t smlInsertData(SSmlHandle *info) { code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%" PRIx64 " catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName); + taosMemoryFree(measure); return code; } taosHashPut(info->pVgHash, (const char *)&vg.vgId, sizeof(vg.vgId), (char *)&vg, sizeof(vg)); @@ -1426,6 +1449,7 @@ static int32_t smlInsertData(SSmlHandle *info) { (SSmlSTableMeta **)taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen); if (unlikely(NULL == pMeta || NULL == (*pMeta)->tableMeta)) { uError("SML:0x%" PRIx64 " NULL == pMeta. table name: %s", info->id, tableData->childTableName); + taosMemoryFree(measure); return TSDB_CODE_SML_INTERNAL_ERROR; } @@ -1435,11 +1459,6 @@ static int32_t smlInsertData(SSmlHandle *info) { uDebug("SML:0x%" PRIx64 " smlInsertData table:%s, uid:%" PRIu64 ", format:%d", info->id, pName.tname, tableData->uid, info->dataFormat); - int measureLen = tableData->sTableNameLen; - char *measure = (char *)taosMemoryMalloc(tableData->sTableNameLen); - memcpy(measure, tableData->sTableName, tableData->sTableNameLen); - PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen); - code = smlBindData(info->pQuery, info->dataFormat, tableData->tags, (*pMeta)->cols, tableData->cols, (*pMeta)->tableMeta, tableData->childTableName, measure, measureLen, info->ttl, info->msgBuf.buf, info->msgBuf.len); diff --git a/source/client/src/clientSmlJson.c b/source/client/src/clientSmlJson.c index 9683d6799a..76794fd187 100644 --- a/source/client/src/clientSmlJson.c +++ b/source/client/src/clientSmlJson.c @@ -996,8 +996,8 @@ static int32_t smlParseJSONStringExt(SSmlHandle *info, cJSON *root, SSmlLineInfo uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id); return TSDB_CODE_INVALID_TIMESTAMP; } - SSmlKv kvTs = {.key = TS, - .keyLen = TS_LEN, + SSmlKv kvTs = {.key = tsSmlTsDefaultName, + .keyLen = strlen(tsSmlTsDefaultName), .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes}; @@ -1200,8 +1200,8 @@ static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo * return TSDB_CODE_INVALID_TIMESTAMP; } } - SSmlKv kvTs = {.key = TS, - .keyLen = TS_LEN, + SSmlKv kvTs = {.key = tsSmlTsDefaultName, + .keyLen = strlen(tsSmlTsDefaultName), .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes}; diff --git a/source/client/src/clientSmlLine.c b/source/client/src/clientSmlLine.c index c5832ce1ce..1ee2cfbedf 100644 --- a/source/client/src/clientSmlLine.c +++ b/source/client/src/clientSmlLine.c @@ -157,6 +157,7 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin measure = (char *)taosMemoryMalloc(currElement->measureLen); memcpy(measure, currElement->measure, currElement->measureLen); PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen); + smlStrReplace(measure, measureLen); } STableMeta *pTableMeta = smlGetMeta(info, measure, measureLen); if (currElement->measureEscaped) { @@ -365,6 +366,7 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin measure = (char *)taosMemoryMalloc(currElement->measureLen); memcpy(measure, currElement->measure, currElement->measureLen); PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen); + smlStrReplace(measure, measureLen); } STableMeta *pTableMeta = smlGetMeta(info, measure, measureLen); if (currElement->measureEscaped) { @@ -651,8 +653,8 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine return TSDB_CODE_INVALID_TIMESTAMP; } // add ts to - SSmlKv kv = {.key = TS, - .keyLen = TS_LEN, + SSmlKv kv = {.key = tsSmlTsDefaultName, + .keyLen = strlen(tsSmlTsDefaultName), .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes, diff --git a/source/client/src/clientSmlTelnet.c b/source/client/src/clientSmlTelnet.c index b3f45a3107..c378f9b1c3 100644 --- a/source/client/src/clientSmlTelnet.c +++ b/source/client/src/clientSmlTelnet.c @@ -260,8 +260,8 @@ int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql); return TSDB_CODE_INVALID_TIMESTAMP; } - SSmlKv kvTs = {.key = TS, - .keyLen = TS_LEN, + SSmlKv kvTs = {.key = tsSmlTsDefaultName, + .keyLen = strlen(tsSmlTsDefaultName), .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes}; diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 92dd4ea419..ae82be2470 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -586,30 +586,36 @@ static int32_t asyncCommitOffset(tmq_t* tmq, char* pTopicName, int32_t vgId, STq if(code != 0){ goto end; } - if (offsetVal->type > 0 && !tOffsetEqual(offsetVal, &pVg->offsetInfo.committedOffset)) { - char offsetBuf[TSDB_OFFSET_LEN] = {0}; - tFormatOffset(offsetBuf, tListLen(offsetBuf), offsetVal); - - char commitBuf[TSDB_OFFSET_LEN] = {0}; - tFormatOffset(commitBuf, tListLen(commitBuf), &pVg->offsetInfo.committedOffset); - - SMqCommitCbParamSet* pParamSet = prepareCommitCbParamSet(tmq, pCommitFp, userParam, 0); - if (pParamSet == NULL) { - code = TSDB_CODE_OUT_OF_MEMORY; - goto end; - } - code = doSendCommitMsg(tmq, pVg->vgId, &pVg->epSet, offsetVal, pTopicName, pParamSet); - if (code != TSDB_CODE_SUCCESS) { - tscError("consumer:0x%" PRIx64 " topic:%s on vgId:%d end commit msg failed, send offset:%s committed:%s, code:%s", - tmq->consumerId, pTopicName, pVg->vgId, offsetBuf, commitBuf, tstrerror(terrno)); - taosMemoryFree(pParamSet); - goto end; - } - - tscInfo("consumer:0x%" PRIx64 " topic:%s on vgId:%d send commit msg success, send offset:%s committed:%s", - tmq->consumerId, pTopicName, pVg->vgId, offsetBuf, commitBuf); - pVg->offsetInfo.committedOffset = *offsetVal; + if (offsetVal->type <= 0) { + code = TSDB_CODE_TMQ_INVALID_MSG; + goto end; } + if (tOffsetEqual(offsetVal, &pVg->offsetInfo.committedOffset)){ + code = TSDB_CODE_TMQ_SAME_COMMITTED_VALUE; + goto end; + } + char offsetBuf[TSDB_OFFSET_LEN] = {0}; + tFormatOffset(offsetBuf, tListLen(offsetBuf), offsetVal); + + char commitBuf[TSDB_OFFSET_LEN] = {0}; + tFormatOffset(commitBuf, tListLen(commitBuf), &pVg->offsetInfo.committedOffset); + + SMqCommitCbParamSet* pParamSet = prepareCommitCbParamSet(tmq, pCommitFp, userParam, 0); + if (pParamSet == NULL) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto end; + } + code = doSendCommitMsg(tmq, pVg->vgId, &pVg->epSet, offsetVal, pTopicName, pParamSet); + if (code != TSDB_CODE_SUCCESS) { + tscError("consumer:0x%" PRIx64 " topic:%s on vgId:%d end commit msg failed, send offset:%s committed:%s, code:%s", + tmq->consumerId, pTopicName, pVg->vgId, offsetBuf, commitBuf, tstrerror(terrno)); + taosMemoryFree(pParamSet); + goto end; + } + + tscInfo("consumer:0x%" PRIx64 " topic:%s on vgId:%d send commit msg success, send offset:%s committed:%s", + tmq->consumerId, pTopicName, pVg->vgId, offsetBuf, commitBuf); + pVg->offsetInfo.committedOffset = *offsetVal; end: taosRUnLockLatch(&tmq->lock); @@ -650,7 +656,8 @@ static void asyncCommitFromResult(tmq_t* tmq, const TAOS_RES* pRes, tmq_commit_c code = asyncCommitOffset(tmq, pTopicName, vgId, &offsetVal, pCommitFp, userParam); end: - if(code != TSDB_CODE_SUCCESS){ + if(code != TSDB_CODE_SUCCESS && pCommitFp != NULL){ + if(code == TSDB_CODE_TMQ_SAME_COMMITTED_VALUE) code = TSDB_CODE_SUCCESS; pCommitFp(tmq, code, userParam); } } @@ -1859,8 +1866,8 @@ static int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* p static void updateVgInfo(SMqClientVg* pVg, STqOffsetVal* reqOffset, STqOffsetVal* rspOffset, int64_t sver, int64_t ever, int64_t consumerId){ if (!pVg->seekUpdated) { tscDebug("consumer:0x%" PRIx64" local offset is update, since seekupdate not set", consumerId); - if(reqOffset->type != 0) pVg->offsetInfo.beginOffset = *reqOffset; - if(rspOffset->type != 0) pVg->offsetInfo.endOffset = *rspOffset; + pVg->offsetInfo.beginOffset = *reqOffset; + pVg->offsetInfo.endOffset = *rspOffset; } else { tscDebug("consumer:0x%" PRIx64" local offset is NOT update, since seekupdate is set", consumerId); } @@ -1948,7 +1955,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) { } taosWUnLockLatch(&tmq->lock); } else { - tscDebug("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", + tscInfo("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", tmq->consumerId, pollRspWrapper->vgId, pDataRsp->head.epoch, consumerEpoch); pRspWrapper = tmqFreeRspWrapper(pRspWrapper); taosFreeQitem(pollRspWrapper); @@ -1979,7 +1986,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) { taosWUnLockLatch(&tmq->lock); return pRsp; } else { - tscDebug("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", + tscInfo("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", tmq->consumerId, pollRspWrapper->vgId, pollRspWrapper->metaRsp.head.epoch, consumerEpoch); pRspWrapper = tmqFreeRspWrapper(pRspWrapper); taosFreeQitem(pollRspWrapper); @@ -2036,7 +2043,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) { taosWUnLockLatch(&tmq->lock); return pRsp; } else { - tscDebug("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", + tscInfo("consumer:0x%" PRIx64 " vgId:%d msg discard since epoch mismatch: msg epoch %d, consumer epoch %d", tmq->consumerId, pollRspWrapper->vgId, pollRspWrapper->taosxRsp.head.epoch, consumerEpoch); pRspWrapper = tmqFreeRspWrapper(pRspWrapper); taosFreeQitem(pollRspWrapper); @@ -2350,7 +2357,7 @@ int32_t tmq_commit_sync(tmq_t* tmq, const TAOS_RES* pRes) { tsem_destroy(&pInfo->sem); taosMemoryFree(pInfo); - tscDebug("consumer:0x%" PRIx64 " sync commit done, code:%s", tmq->consumerId, tstrerror(code)); + tscInfo("consumer:0x%" PRIx64 " sync res commit done, code:%s", tmq->consumerId, tstrerror(code)); return code; } @@ -2406,15 +2413,17 @@ int32_t tmq_commit_offset_sync(tmq_t *tmq, const char *pTopicName, int32_t vgId, tsem_init(&pInfo->sem, 0, 0); pInfo->code = 0; - asyncCommitOffset(tmq, tname, vgId, &offsetVal, commitCallBackFn, pInfo); - - tsem_wait(&pInfo->sem); - code = pInfo->code; + code = asyncCommitOffset(tmq, tname, vgId, &offsetVal, commitCallBackFn, pInfo); + if(code == 0){ + tsem_wait(&pInfo->sem); + code = pInfo->code; + } + if(code == TSDB_CODE_TMQ_SAME_COMMITTED_VALUE) code = TSDB_CODE_SUCCESS; tsem_destroy(&pInfo->sem); taosMemoryFree(pInfo); - tscInfo("consumer:0x%" PRIx64 " sync send seek to vgId:%d, offset:%" PRId64" code:%s", tmq->consumerId, vgId, offset, tstrerror(code)); + tscInfo("consumer:0x%" PRIx64 " sync send commit to vgId:%d, offset:%" PRId64" code:%s", tmq->consumerId, vgId, offset, tstrerror(code)); return code; } @@ -2451,10 +2460,11 @@ void tmq_commit_offset_async(tmq_t *tmq, const char *pTopicName, int32_t vgId, i code = asyncCommitOffset(tmq, tname, vgId, &offsetVal, cb, param); - tscInfo("consumer:0x%" PRIx64 " async send seek to vgId:%d, offset:%" PRId64" code:%s", tmq->consumerId, vgId, offset, tstrerror(code)); + tscInfo("consumer:0x%" PRIx64 " async send commit to vgId:%d, offset:%" PRId64" code:%s", tmq->consumerId, vgId, offset, tstrerror(code)); end: if(code != 0 && cb != NULL){ + if(code == TSDB_CODE_TMQ_SAME_COMMITTED_VALUE) code = TSDB_CODE_SUCCESS; cb(tmq, code, param); } } diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index f06eeb230f..0546ed7f47 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -105,6 +105,8 @@ char *tsClientCrashReportUri = "/ccrashreport"; char *tsSvrCrashReportUri = "/dcrashreport"; // schemaless +bool tsSmlDot2Underline = true; +char tsSmlTsDefaultName[TSDB_COL_NAME_LEN] = "_ts"; char tsSmlTagName[TSDB_COL_NAME_LEN] = "_tag_null"; char tsSmlChildTableName[TSDB_TABLE_NAME_LEN] = ""; // user defined child table name can be specified in tag value. // If set to empty system will generate table name using MD5 hash. @@ -366,6 +368,8 @@ static int32_t taosAddClientCfg(SConfig *pCfg) { if (cfgAddBool(pCfg, "keepColumnName", tsKeepColumnName, CFG_SCOPE_CLIENT) != 0) return -1; if (cfgAddString(pCfg, "smlChildTableName", "", CFG_SCOPE_CLIENT) != 0) return -1; if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, CFG_SCOPE_CLIENT) != 0) return -1; + if (cfgAddString(pCfg, "smlTsDefaultName", tsSmlTsDefaultName, CFG_SCOPE_CLIENT) != 0) return -1; + if (cfgAddBool(pCfg, "smlDot2Underline", tsSmlDot2Underline, CFG_SCOPE_CLIENT) != 0) return -1; // if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, CFG_SCOPE_CLIENT) != 0) return -1; // if (cfgAddInt32(pCfg, "smlBatchSize", tsSmlBatchSize, 1, INT32_MAX, CFG_SCOPE_CLIENT) != 0) return -1; if (cfgAddInt32(pCfg, "maxInsertBatchRows", tsMaxInsertBatchRows, 1, INT32_MAX, CFG_SCOPE_CLIENT) != 0) return -1; @@ -801,6 +805,8 @@ static int32_t taosSetClientCfg(SConfig *pCfg) { tstrncpy(tsSmlChildTableName, cfgGetItem(pCfg, "smlChildTableName")->str, TSDB_TABLE_NAME_LEN); tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN); + tstrncpy(tsSmlTsDefaultName, cfgGetItem(pCfg, "smlTsDefaultName")->str, TSDB_COL_NAME_LEN); + tsSmlDot2Underline = cfgGetItem(pCfg, "smlDot2Underline")->bval; // tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval; // tsSmlBatchSize = cfgGetItem(pCfg, "smlBatchSize")->i32; @@ -1243,6 +1249,10 @@ int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) { // tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval; // } else if (strcasecmp("smlBatchSize", name) == 0) { // tsSmlBatchSize = cfgGetItem(pCfg, "smlBatchSize")->i32; + } else if(strcasecmp("smlTsDefaultName", name) == 0) { + tstrncpy(tsSmlTsDefaultName, cfgGetItem(pCfg, "smlTsDefaultName")->str, TSDB_COL_NAME_LEN); + } else if(strcasecmp("smlDot2Underline", name) == 0) { + tsSmlDot2Underline = cfgGetItem(pCfg, "smlDot2Underline")->bval; } else if (strcasecmp("shellActivityTimer", name) == 0) { tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32; } else if (strcasecmp("supportVnodes", name) == 0) { diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 73a0bed48f..ef0006e7ab 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -7227,6 +7227,9 @@ bool tOffsetEqual(const STqOffsetVal *pLeft, const STqOffsetVal *pRight) { return pLeft->uid == pRight->uid && pLeft->ts == pRight->ts; } else if (pLeft->type == TMQ_OFFSET__SNAPSHOT_META) { return pLeft->uid == pRight->uid; + } else { + uError("offset type:%d", pLeft->type); + ASSERT(0); } } return false; diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 6081b9a530..82492f930e 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -94,7 +94,7 @@ void mndDropConsumerFromSdb(SMnode *pMnode, int64_t consumerId){ bool mndRebTryStart() { int32_t old = atomic_val_compare_exchange_32(&mqRebInExecCnt, 0, 1); - mInfo("tq timer, rebalance counter old val:%d", old); + mDebug("tq timer, rebalance counter old val:%d", old); return old == 0; } @@ -116,7 +116,7 @@ void mndRebCntDec() { int32_t newVal = val - 1; int32_t oldVal = atomic_val_compare_exchange_32(&mqRebInExecCnt, val, newVal); if (oldVal == val) { - mInfo("rebalance trans end, rebalance counter:%d", newVal); + mDebug("rebalance trans end, rebalance counter:%d", newVal); break; } } @@ -420,6 +420,7 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) { SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, data->topicName); if(pSub == NULL){ + ASSERT(0); continue; } taosWLockLatch(&pSub->lock); @@ -515,7 +516,10 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) { char *topic = taosArrayGetP(pConsumer->currentTopics, i); SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, topic); // txn guarantees pSub is created - if(pSub == NULL) continue; + if(pSub == NULL) { + ASSERT(0); + continue; + } taosRLockLatch(&pSub->lock); SMqSubTopicEp topicEp = {0}; @@ -524,6 +528,7 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) { // 2.1 fetch topic schema SMqTopicObj *pTopic = mndAcquireTopic(pMnode, topic); if(pTopic == NULL) { + ASSERT(0); taosRUnLockLatch(&pSub->lock); mndReleaseSubscribe(pMnode, pSub); continue; @@ -898,7 +903,7 @@ static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) { mInfo("consumer:0x%" PRIx64 " sub insert, cgroup:%s status:%d(%s) epoch:%d", pConsumer->consumerId, pConsumer->cgroup, pConsumer->status, mndConsumerStatusName(pConsumer->status), pConsumer->epoch); - pConsumer->subscribeTime = taosGetTimestampMs(); + pConsumer->subscribeTime = pConsumer->createTime; return 0; } diff --git a/source/dnode/mnode/impl/src/mndShow.c b/source/dnode/mnode/impl/src/mndShow.c index c50b205f37..44f4751700 100644 --- a/source/dnode/mnode/impl/src/mndShow.c +++ b/source/dnode/mnode/impl/src/mndShow.c @@ -20,6 +20,7 @@ #define SHOW_STEP_SIZE 100 #define SHOW_COLS_STEP_SIZE 4096 +#define SHOW_PRIVILEGES_STEP_SIZE 2048 static SShowObj *mndCreateShowObj(SMnode *pMnode, SRetrieveTableReq *pReq); static void mndFreeShowObj(SShowObj *pShow); @@ -234,6 +235,8 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) { if(pShow->type == TSDB_MGMT_TABLE_COL){ // expend capacity for ins_columns rowsToRead = SHOW_COLS_STEP_SIZE; + } else if (pShow->type == TSDB_MGMT_TABLE_PRIVILEGES) { + rowsToRead = SHOW_PRIVILEGES_STEP_SIZE; } ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type]; if (retrieveFp == NULL) { diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index 85054e5cd7..6f50b9ff9f 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -1169,7 +1169,7 @@ static int32_t buildResult(SSDataBlock *pBlock, int32_t* numOfRows, int64_t cons pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, *numOfRows, (const char *)consumerIdHex, consumerId == -1); - mDebug("mnd show subscriptions: topic %s, consumer:0x%" PRIx64 " cgroup %s vgid %d", varDataVal(topic), + mInfo("mnd show subscriptions: topic %s, consumer:0x%" PRIx64 " cgroup %s vgid %d", varDataVal(topic), consumerId, varDataVal(cgroup), pVgEp->vgId); // offset diff --git a/source/dnode/mnode/impl/src/mndUser.c b/source/dnode/mnode/impl/src/mndUser.c index cbde56a860..65c1cfbea2 100644 --- a/source/dnode/mnode/impl/src/mndUser.c +++ b/source/dnode/mnode/impl/src/mndUser.c @@ -1174,26 +1174,30 @@ static void mndLoopHash(SHashObj *hash, char *priType, SSDataBlock *pBlock, int3 if (strcmp("t", value) != 0) { SNode *pAst = NULL; int32_t sqlLen = 0; - char sql[TSDB_EXPLAIN_RESULT_ROW_SIZE] = {0}; + size_t bufSz = strlen(value) + 1; + char* sql = taosMemoryMalloc(bufSz + 1); + char* obj = taosMemoryMalloc(TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE); - if (nodesStringToNode(value, &pAst) == 0) { - nodesNodeToSQL(pAst, sql, TSDB_EXPLAIN_RESULT_ROW_SIZE, &sqlLen); + if (sql != NULL && obj != NULL && nodesStringToNode(value, &pAst) == 0) { + nodesNodeToSQL(pAst, sql, bufSz, &sqlLen); nodesDestroyNode(pAst); } else { sqlLen = 5; sprintf(sql, "error"); } - char obj[TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE] = {0}; STR_WITH_MAXSIZE_TO_VARSTR(obj, sql, pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, *numOfRows, (const char *)obj, false); + taosMemoryFree(obj); + taosMemoryFree(sql); } else { - char condition[TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE] = {0}; + char* condition = taosMemoryMalloc(TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE); STR_WITH_MAXSIZE_TO_VARSTR(condition, "", pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, *numOfRows, (const char *)condition, false); + taosMemoryFree(condition); } (*numOfRows)++; @@ -1209,16 +1213,34 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock int32_t cols = 0; char *pWrite; + bool fetchNextUser = pShow->restore ? false : true; + pShow->restore = false; + while (numOfRows < rows) { - pShow->pIter = sdbFetch(pSdb, SDB_USER, pShow->pIter, (void **)&pUser); - if (pShow->pIter == NULL) break; + if (fetchNextUser) { + pShow->pIter = sdbFetch(pSdb, SDB_USER, pShow->pIter, (void **)&pUser); + if (pShow->pIter == NULL) break; + } else { + fetchNextUser = true; + void *pKey = taosHashGetKey(pShow->pIter, NULL); + pUser = sdbAcquire(pSdb, SDB_USER, pKey); + if (!pUser) { + continue; + } + } int32_t numOfReadDbs = taosHashGetSize(pUser->readDbs); int32_t numOfWriteDbs = taosHashGetSize(pUser->writeDbs); int32_t numOfTopics = taosHashGetSize(pUser->topics); int32_t numOfReadTbs = taosHashGetSize(pUser->readTbs); int32_t numOfWriteTbs = taosHashGetSize(pUser->writeTbs); - if (numOfRows + numOfReadDbs + numOfWriteDbs + numOfTopics + numOfReadTbs + numOfWriteTbs >= rows) break; + if (numOfRows + numOfReadDbs + numOfWriteDbs + numOfTopics + numOfReadTbs + numOfWriteTbs >= rows) { + mInfo("will restore. current num of rows: %d, read dbs %d, write dbs %d, topics %d, read tables %d, write tables %d", + numOfRows, numOfReadDbs, numOfWriteDbs, numOfTopics, numOfReadTbs, numOfWriteTbs); + pShow->restore = true; + sdbRelease(pSdb, pUser); + break; + } if (pUser->superUser) { cols = 0; @@ -1242,10 +1264,11 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)tableName, false); - char condition[TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE] = {0}; + char* condition = taosMemoryMalloc(TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE); STR_WITH_MAXSIZE_TO_VARSTR(condition, "", pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)condition, false); + taosMemoryFree(condition); numOfRows++; } @@ -1276,10 +1299,11 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)tableName, false); - char condition[TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE] = {0}; + char* condition = taosMemoryMalloc(TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE); STR_WITH_MAXSIZE_TO_VARSTR(condition, "", pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)condition, false); + taosMemoryFree(condition); numOfRows++; db = taosHashIterate(pUser->readDbs, db); @@ -1311,10 +1335,11 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)tableName, false); - char condition[TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE] = {0}; + char* condition = taosMemoryMalloc(TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE); STR_WITH_MAXSIZE_TO_VARSTR(condition, "", pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)condition, false); + taosMemoryFree(condition); numOfRows++; db = taosHashIterate(pUser->writeDbs, db); @@ -1348,10 +1373,11 @@ static int32_t mndRetrievePrivileges(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)tableName, false); - char condition[TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE] = {0}; + char* condition = taosMemoryMalloc(TSDB_PRIVILEDGE_CONDITION_LEN + VARSTR_HEADER_SIZE); STR_WITH_MAXSIZE_TO_VARSTR(condition, "", pShow->pMeta->pSchemas[cols].bytes); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)condition, false); + taosMemoryFree(condition); numOfRows++; topic = taosHashIterate(pUser->topics, topic); diff --git a/source/dnode/snode/src/snodeInitApi.c b/source/dnode/snode/src/snodeInitApi.c index c046505630..e737e3fa37 100644 --- a/source/dnode/snode/src/snodeInitApi.c +++ b/source/dnode/snode/src/snodeInitApi.c @@ -78,6 +78,8 @@ void initStateStoreAPI(SStateStore* pStore) { pStore->updateInfoIsUpdated = updateInfoIsUpdated; pStore->updateInfoIsTableInserted = updateInfoIsTableInserted; pStore->updateInfoDestroy = updateInfoDestroy; + pStore->windowSBfDelete = windowSBfDelete; + pStore->windowSBfAdd = windowSBfAdd; pStore->updateInfoInitP = updateInfoInitP; pStore->updateInfoAddCloseWindowSBF = updateInfoAddCloseWindowSBF; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 6f7410bed6..4d433042ad 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -146,6 +146,20 @@ void tqClose(STQ* pTq) { return; } + void* pIter = taosHashIterate(pTq->pPushMgr, NULL); + while (pIter) { + STqHandle* pHandle = *(STqHandle**)pIter; + int32_t vgId = TD_VID(pTq->pVnode); + + if(pHandle->msg != NULL) { + tqPushEmptyDataRsp(pHandle, vgId); + rpcFreeCont(pHandle->msg->pCont); + taosMemoryFree(pHandle->msg); + pHandle->msg = NULL; + } + pIter = taosHashIterate(pTq->pPushMgr, pIter); + } + tqOffsetClose(pTq->pOffsetStore); taosHashCleanup(pTq->pHandle); taosHashCleanup(pTq->pPushMgr); @@ -278,6 +292,10 @@ int32_t tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId) { tqInitDataRsp(&dataRsp, &req); dataRsp.blockNum = 0; dataRsp.rspOffset = dataRsp.reqOffset; + char buf[TSDB_OFFSET_LEN] = {0}; + tFormatOffset(buf, TSDB_OFFSET_LEN, &dataRsp.reqOffset); + tqInfo("tqPushEmptyDataRsp to consumer:0x%"PRIx64 " vgId:%d, offset:%s, reqId:0x%" PRIx64, req.consumerId, vgId, buf, req.reqId); + tqSendDataRsp(pHandle, pHandle->msg, &req, &dataRsp, TMQ_MSG_TYPE__POLL_DATA_RSP, vgId); tDeleteMqDataRsp(&dataRsp); return 0; @@ -373,6 +391,7 @@ int32_t tqProcessSeekReq(STQ* pTq, SRpcMsg* pMsg) { } tqDebug("tmq seek: consumer:0x%" PRIx64 " vgId:%d, subkey %s", req.consumerId, vgId, req.subKey); + STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey)); if (pHandle == NULL) { tqWarn("tmq seek: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", req.consumerId, vgId, req.subKey); @@ -515,10 +534,11 @@ int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg) { while (pIter) { STqHandle* pHandle = *(STqHandle**)pIter; - tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId); + tqInfo("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId); if (ASSERT(pHandle->msg != NULL)) { tqError("pHandle->msg should not be null"); + taosHashCancelIterate(pTq->pPushMgr, pIter); break; }else{ SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME, .pCont = pHandle->msg->pCont, .contLen = pHandle->msg->contLen, .info = pHandle->msg->info}; @@ -556,10 +576,18 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) { // 1. find handle pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey)); if (pHandle == NULL) { - tqError("tmq poll: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", consumerId, vgId, req.subKey); - terrno = TSDB_CODE_INVALID_MSG; - taosWUnLockLatch(&pTq->lock); - return -1; + do{ + if (tqMetaGetHandle(pTq, req.subKey) == 0){ + pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey)); + if(pHandle != NULL){ + break; + } + } + tqError("tmq poll: consumer:0x%" PRIx64 " vgId:%d subkey %s not found", consumerId, vgId, req.subKey); + terrno = TSDB_CODE_INVALID_MSG; + taosWUnLockLatch(&pTq->lock); + return -1; + }while(0); } // 2. check re-balance status @@ -849,30 +877,28 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg taosWLockLatch(&pTq->lock); if (pHandle->consumerId == req.newConsumerId) { // do nothing - tqInfo("vgId:%d consumer:0x%" PRIx64 " remains, no switch occurs, should not reach here", req.vgId, - req.newConsumerId); + tqInfo("vgId:%d no switch consumer:0x%" PRIx64 " remains, because redo wal log", req.vgId, req.newConsumerId); } else { - tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId, - req.newConsumerId); + tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId, req.newConsumerId); atomic_store_64(&pHandle->consumerId, req.newConsumerId); + // atomic_add_fetch_32(&pHandle->epoch, 1); + + // kill executing task + // if(tqIsHandleExec(pHandle)) { + // qTaskInfo_t pTaskInfo = pHandle->execHandle.task; + // if (pTaskInfo != NULL) { + // qKillTask(pTaskInfo, TSDB_CODE_SUCCESS); + // } + + // if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { + // qStreamCloseTsdbReader(pTaskInfo); + // } + // } + // remove if it has been register in the push manager, and return one empty block to consumer + tqUnregisterPushHandle(pTq, pHandle); + ret = tqMetaSaveHandle(pTq, req.subKey, pHandle); } - // atomic_add_fetch_32(&pHandle->epoch, 1); - - // kill executing task - // if(tqIsHandleExec(pHandle)) { - // qTaskInfo_t pTaskInfo = pHandle->execHandle.task; - // if (pTaskInfo != NULL) { - // qKillTask(pTaskInfo, TSDB_CODE_SUCCESS); - // } - - // if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { - // qStreamCloseTsdbReader(pTaskInfo); - // } - // } - // remove if it has been register in the push manager, and return one empty block to consumer - tqUnregisterPushHandle(pTq, pHandle); taosWUnLockLatch(&pTq->lock); - ret = tqMetaSaveHandle(pTq, req.subKey, pHandle); } end: diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index df1c9ca7c9..08019f8a76 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -338,7 +338,7 @@ static int buildHandle(STQ* pTq, STqHandle* handle){ taosArrayDestroy(tbUidList); return -1; } - tqDebug("vgId:%d, tq try to get ctb for stb subscribe, suid:%" PRId64, pVnode->config.vgId, handle->execHandle.execTb.suid); + tqInfo("vgId:%d, tq try to get ctb for stb subscribe, suid:%" PRId64, pVnode->config.vgId, handle->execHandle.execTb.suid); handle->execHandle.pTqReader = tqReaderOpen(pVnode); tqReaderSetTbUidList(handle->execHandle.pTqReader, tbUidList, NULL); taosArrayDestroy(tbUidList); @@ -356,7 +356,7 @@ static int restoreHandle(STQ* pTq, void* pVal, int vLen, STqHandle* handle){ if(buildHandle(pTq, handle) < 0){ return -1; } - tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle->subKey, handle->consumerId, vgId); + tqInfo("tq restore %s consumer %" PRId64 " vgId:%d", handle->subKey, handle->consumerId, vgId); return taosHashPut(pTq->pHandle, handle->subKey, strlen(handle->subKey), handle, sizeof(STqHandle)); } @@ -384,7 +384,7 @@ int32_t tqCreateHandle(STQ* pTq, SMqRebVgReq* req, STqHandle* handle){ if(buildHandle(pTq, handle) < 0){ return -1; } - tqDebug("tq restore %s consumer %" PRId64 " vgId:%d", handle->subKey, handle->consumerId, vgId); + tqInfo("tq restore %s consumer %" PRId64 " vgId:%d", handle->subKey, handle->consumerId, vgId); return taosHashPut(pTq->pHandle, handle->subKey, strlen(handle->subKey), handle, sizeof(STqHandle)); } diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index 5ccf4c825b..a236b98614 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -78,12 +78,12 @@ int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) { memcpy(pHandle->msg->pCont, pMsg->pCont, pMsg->contLen); pHandle->msg->contLen = pMsg->contLen; int32_t ret = taosHashPut(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey), &pHandle, POINTER_BYTES); - tqDebug("vgId:%d data is over, ret:%d, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, ret, + tqInfo("vgId:%d data is over, ret:%d, consumerId:0x%" PRIx64 ", register to pHandle:%p, pCont:%p, len:%d", vgId, ret, pHandle->consumerId, pHandle, pHandle->msg->pCont, pHandle->msg->contLen); return 0; } -int32_t tqUnregisterPushHandle(STQ* pTq, void *handle) { +int tqUnregisterPushHandle(STQ* pTq, void *handle) { STqHandle *pHandle = (STqHandle*)handle; int32_t vgId = TD_VID(pTq->pVnode); @@ -91,7 +91,7 @@ int32_t tqUnregisterPushHandle(STQ* pTq, void *handle) { return 0; } int32_t ret = taosHashRemove(pTq->pPushMgr, pHandle->subKey, strlen(pHandle->subKey)); - tqDebug("vgId:%d remove pHandle:%p,ret:%d consumer Id:0x%" PRIx64, vgId, pHandle, ret, pHandle->consumerId); + tqInfo("vgId:%d remove pHandle:%p,ret:%d consumer Id:0x%" PRIx64, vgId, pHandle, ret, pHandle->consumerId); if(pHandle->msg != NULL) { // tqPushDataRsp(pHandle, vgId); diff --git a/source/dnode/vnode/src/tq/tqUtil.c b/source/dnode/vnode/src/tq/tqUtil.c index 9062084246..55a1cecafe 100644 --- a/source/dnode/vnode/src/tq/tqUtil.c +++ b/source/dnode/vnode/src/tq/tqUtil.c @@ -317,6 +317,7 @@ int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequ // the offset value can not be monotonious increase?? offset = reqOffset; } else { + uError("req offset type is 0"); return TSDB_CODE_TMQ_INVALID_MSG; } diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index 32429ff8c6..a5b5dea505 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -1667,10 +1667,6 @@ static int32_t loadTombFromBlk(const TTombBlkArray *pTombBlkArray, SCacheRowsRea } if (record.version <= pReader->info.verRange.maxVer) { - /* - tsdbError("tomb xx load/cache: vgId:%d fid:%d commit %" PRId64 "~%" PRId64 "~%" PRId64 " tomb records", - TD_VID(pReader->pTsdb->pVnode), pReader->pCurFileSet->fid, record.skey, record.ekey, uid); - */ SDelData delData = {.version = record.version, .sKey = record.skey, .eKey = record.ekey}; taosArrayPush(pInfo->pTombData, &delData); } @@ -1912,15 +1908,14 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie int indexSize = TARRAY_SIZE(state->pIndexList); if (indexSize <= 0) { - clearLastFileSet(state); - state->state = SFSNEXTROW_FILESET; - goto _next_fileset; + goto _check_stt_data; } state->state = SFSNEXTROW_INDEXLIST; state->iBrinIndex = indexSize; } + _check_stt_data: if (state->pFileSet != state->pr->pCurFileSet) { state->pr->pCurFileSet = state->pFileSet; } diff --git a/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c b/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c index 9d07bcb446..dc5e3649cc 100644 --- a/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c +++ b/source/dnode/vnode/src/tsdb/tsdbDataFileRW.c @@ -1229,11 +1229,16 @@ static int32_t tsdbDataFileDoWriteTombRecord(SDataFileWriter *writer, const STom int32_t c = tTombRecordCompare(record, record1); if (c < 0) { - break; + goto _write; } else if (c > 0) { code = tTombBlockPut(writer->tombBlock, record1); TSDB_CHECK_CODE(code, lino, _exit); + tsdbTrace("vgId:%d write tomb record to tomb file:%s, cid:%" PRId64 ", suid:%" PRId64 ", uid:%" PRId64 + ", version:%" PRId64, + TD_VID(writer->config->tsdb->pVnode), writer->fd[TSDB_FTYPE_TOMB]->path, writer->config->cid, + record1->suid, record1->uid, record1->version); + if (TOMB_BLOCK_SIZE(writer->tombBlock) >= writer->config->maxRow) { code = tsdbDataFileDoWriteTombBlock(writer); TSDB_CHECK_CODE(code, lino, _exit); @@ -1266,6 +1271,11 @@ _write: code = tTombBlockPut(writer->tombBlock, record); TSDB_CHECK_CODE(code, lino, _exit); + tsdbTrace("vgId:%d write tomb record to tomb file:%s, cid:%" PRId64 ", suid:%" PRId64 ", uid:%" PRId64 + ", version:%" PRId64, + TD_VID(writer->config->tsdb->pVnode), writer->fd[TSDB_FTYPE_TOMB]->path, writer->config->cid, record->suid, + record->uid, record->version); + if (TOMB_BLOCK_SIZE(writer->tombBlock) >= writer->config->maxRow) { code = tsdbDataFileDoWriteTombBlock(writer); TSDB_CHECK_CODE(code, lino, _exit); diff --git a/source/dnode/vnode/src/tsdb/tsdbFS2.c b/source/dnode/vnode/src/tsdb/tsdbFS2.c index 1a0104945f..6e7595c6ef 100644 --- a/source/dnode/vnode/src/tsdb/tsdbFS2.c +++ b/source/dnode/vnode/src/tsdb/tsdbFS2.c @@ -537,6 +537,9 @@ static void tsdbDoWaitBgTask(STFileSystem *fs, STFSBgTask *task) { if (task->numWait == 0) { taosThreadCondDestroy(task->done); + if (task->free) { + task->free(task->arg); + } taosMemoryFree(task); } } @@ -546,6 +549,9 @@ static void tsdbDoDoneBgTask(STFileSystem *fs, STFSBgTask *task) { taosThreadCondBroadcast(task->done); } else { taosThreadCondDestroy(task->done); + if (task->free) { + task->free(task->arg); + } taosMemoryFree(task); } } @@ -627,7 +633,7 @@ int32_t tsdbFSEditCommit(STFileSystem *fs) { SSttLvl *lvl = TARRAY2_FIRST(fset->lvlArr); if (lvl->level != 0 || TARRAY2_SIZE(lvl->fobjArr) < fs->tsdb->pVnode->config.sttTrigger) continue; - code = tsdbFSScheduleBgTask(fs, TSDB_BG_TASK_MERGER, tsdbMerge, fs->tsdb, NULL); + code = tsdbFSScheduleBgTask(fs, TSDB_BG_TASK_MERGER, tsdbMerge, NULL, fs->tsdb, NULL); TSDB_CHECK_CODE(code, lino, _exit); break; @@ -774,8 +780,8 @@ static int32_t tsdbFSRunBgTask(void *arg) { return 0; } -static int32_t tsdbFSScheduleBgTaskImpl(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void *arg, - int64_t *taskid) { +static int32_t tsdbFSScheduleBgTaskImpl(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void (*free)(void *), + void *arg, int64_t *taskid) { if (fs->stop) { return 0; // TODO: use a better error code } @@ -798,6 +804,7 @@ static int32_t tsdbFSScheduleBgTaskImpl(STFileSystem *fs, EFSBgTaskT type, int32 task->type = type; task->run = run; + task->free = free; task->arg = arg; task->scheduleTime = taosGetTimestampMs(); task->taskid = ++fs->taskid; @@ -819,9 +826,10 @@ static int32_t tsdbFSScheduleBgTaskImpl(STFileSystem *fs, EFSBgTaskT type, int32 return 0; } -int32_t tsdbFSScheduleBgTask(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void *arg, int64_t *taskid) { +int32_t tsdbFSScheduleBgTask(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void (*free)(void *), void *arg, + int64_t *taskid) { taosThreadMutexLock(fs->mutex); - int32_t code = tsdbFSScheduleBgTaskImpl(fs, type, run, arg, taskid); + int32_t code = tsdbFSScheduleBgTaskImpl(fs, type, run, free, arg, taskid); taosThreadMutexUnlock(fs->mutex); return code; } diff --git a/source/dnode/vnode/src/tsdb/tsdbFS2.h b/source/dnode/vnode/src/tsdb/tsdbFS2.h index 8270581e58..e814ab2fff 100644 --- a/source/dnode/vnode/src/tsdb/tsdbFS2.h +++ b/source/dnode/vnode/src/tsdb/tsdbFS2.h @@ -59,7 +59,8 @@ int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT e int32_t tsdbFSEditCommit(STFileSystem *fs); int32_t tsdbFSEditAbort(STFileSystem *fs); // background task -int32_t tsdbFSScheduleBgTask(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void *arg, int64_t *taskid); +int32_t tsdbFSScheduleBgTask(STFileSystem *fs, EFSBgTaskT type, int32_t (*run)(void *), void (*free)(void *), void *arg, + int64_t *taskid); int32_t tsdbFSWaitBgTask(STFileSystem *fs, int64_t taskid); int32_t tsdbFSWaitAllBgTask(STFileSystem *fs); int32_t tsdbFSDisableBgTask(STFileSystem *fs); @@ -70,6 +71,7 @@ int32_t tsdbFSGetFSet(STFileSystem *fs, int32_t fid, STFileSet **fset); struct STFSBgTask { EFSBgTaskT type; int32_t (*run)(void *arg); + void (*free)(void *arg); void *arg; TdThreadCond done[1]; diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c index d5d8ba130c..e1756333c5 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead2.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c @@ -1729,45 +1729,41 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader // row in last file block TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex); - int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader); + int64_t ts = getCurrentKeyInLastBlock(pLastBlockReader); + if (ASCENDING_TRAVERSE(pReader->info.order)) { - if (key < tsLast) { + if (key < ts) { // imem, mem are all empty, file blocks (data blocks and last block) exist return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader); - } else if (key > tsLast) { - return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false); - } - } else { - if (key > tsLast) { - return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader); - } else if (key < tsLast) { + } else if (key == ts) { + SRow* pTSRow = NULL; + int32_t code = tsdbRowMergerAdd(pMerger, &fRow, pReader->info.pSchema); + if (code != TSDB_CODE_SUCCESS) { + return code; + } + + doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader); + + TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree); + tsdbRowMergerAdd(pMerger, pRow1, NULL); + + doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, pMerger, &pReader->info.verRange, pReader->idStr); + + code = tsdbRowMergerGetRow(pMerger, &pTSRow); + if (code != TSDB_CODE_SUCCESS) { + return code; + } + + code = doAppendRowFromTSRow(pReader->resBlockInfo.pResBlock, pReader, pTSRow, pBlockScanInfo); + + taosMemoryFree(pTSRow); + tsdbRowMergerClear(pMerger); + return code; + } else { // key > ts return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false); } + } else { // desc order + return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, pBlockData, true); } - // the following for key == tsLast - SRow* pTSRow = NULL; - int32_t code = tsdbRowMergerAdd(pMerger, &fRow, pReader->info.pSchema); - if (code != TSDB_CODE_SUCCESS) { - return code; - } - - doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader); - - TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree); - tsdbRowMergerAdd(pMerger, pRow1, NULL); - - doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, pMerger, &pReader->info.verRange, pReader->idStr); - - code = tsdbRowMergerGetRow(pMerger, &pTSRow); - if (code != TSDB_CODE_SUCCESS) { - return code; - } - - code = doAppendRowFromTSRow(pReader->resBlockInfo.pResBlock, pReader, pTSRow, pBlockScanInfo); - - taosMemoryFree(pTSRow); - tsdbRowMergerClear(pMerger); - return code; - } else { // only last block exists return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false); } @@ -2194,8 +2190,7 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanI SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; TSDBROW *pRow = NULL, *piRow = NULL; - int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : - (ASCENDING_TRAVERSE(pReader->info.order) ? INT64_MAX : INT64_MIN); + int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN; if (pBlockScanInfo->iter.hasVal) { pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader); } @@ -2569,18 +2564,9 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) { // load the last data block of current table STableBlockScanInfo* pScanInfo = *(STableBlockScanInfo**)pStatus->pTableIter; - if (pScanInfo == NULL) { - tsdbError("table Iter is null, invalid pScanInfo, try next table %s", pReader->idStr); - bool hasNexTable = moveToNextTable(pUidList, pStatus); - if (!hasNexTable) { - return TSDB_CODE_SUCCESS; - } - - continue; - } - if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pScanInfo->uid, sizeof(pScanInfo->uid))) { // reset the index in last block when handing a new file + // doCleanupTableScanInfo(pScanInfo); bool hasNexTable = moveToNextTable(pUidList, pStatus); if (!hasNexTable) { return TSDB_CODE_SUCCESS; @@ -2589,6 +2575,9 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) { continue; } + // reset the index in last block when handing a new file + // doCleanupTableScanInfo(pScanInfo); + bool hasDataInLastFile = initLastBlockReader(pLastBlockReader, pScanInfo, pReader); if (!hasDataInLastFile) { bool hasNexTable = moveToNextTable(pUidList, pStatus); @@ -2678,32 +2667,16 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) { (ASCENDING_TRAVERSE(pReader->info.order)) ? pBlockInfo->record.firstKey : pBlockInfo->record.lastKey; code = buildDataBlockFromBuf(pReader, pScanInfo, endKey); } else { - bool bHasDataInLastBlock = hasDataInLastBlock(pLastBlockReader); - int64_t tsLast = bHasDataInLastBlock ? getCurrentKeyInLastBlock(pLastBlockReader) : INT64_MIN; - if (!bHasDataInLastBlock || ((ASCENDING_TRAVERSE(pReader->info.order) && pBlockInfo->record.lastKey < tsLast) || - (!ASCENDING_TRAVERSE(pReader->info.order) && pBlockInfo->record.firstKey > tsLast))) { - // whole block is required, return it directly - SDataBlockInfo* pInfo = &pReader->resBlockInfo.pResBlock->info; - pInfo->rows = pBlockInfo->record.numRow; - pInfo->id.uid = pScanInfo->uid; - pInfo->dataLoad = 0; - pInfo->window = (STimeWindow){.skey = pBlockInfo->record.firstKey, .ekey = pBlockInfo->record.lastKey}; - setComposedBlockFlag(pReader, false); - setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlockInfo->record.lastKey, pReader->info.order); + if (hasDataInLastBlock(pLastBlockReader) && !ASCENDING_TRAVERSE(pReader->info.order)) { + // only return the rows in last block + int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader); + ASSERT(tsLast >= pBlockInfo->record.lastKey); - // update the last key for the corresponding table - pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->info.order) ? pInfo->window.ekey : pInfo->window.skey; - tsdbDebug("%p uid:%" PRIu64 - " clean file block retrieved from file, global index:%d, " - "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s", - pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlockInfo->record.numRow, - pBlockInfo->record.firstKey, pBlockInfo->record.lastKey, pReader->idStr); - } else { SBlockData* pBData = &pReader->status.fileBlockData; tBlockDataReset(pBData); SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock; - tsdbDebug("load data in last block firstly %s", pReader->idStr); + tsdbDebug("load data in last block firstly, due to desc scan data, %s", pReader->idStr); int64_t st = taosGetTimestampUs(); @@ -2734,8 +2707,23 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) { pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey, pResBlock->info.rows, el, pReader->idStr); } - } + } else { // whole block is required, return it directly + SDataBlockInfo* pInfo = &pReader->resBlockInfo.pResBlock->info; + pInfo->rows = pBlockInfo->record.numRow; + pInfo->id.uid = pScanInfo->uid; + pInfo->dataLoad = 0; + pInfo->window = (STimeWindow){.skey = pBlockInfo->record.firstKey, .ekey = pBlockInfo->record.lastKey}; + setComposedBlockFlag(pReader, false); + setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlockInfo->record.lastKey, pReader->info.order); + // update the last key for the corresponding table + pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->info.order) ? pInfo->window.ekey : pInfo->window.skey; + tsdbDebug("%p uid:%" PRIu64 + " clean file block retrieved from file, global index:%d, " + "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s", + pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlockInfo->record.numRow, + pBlockInfo->record.firstKey, pBlockInfo->record.lastKey, pReader->idStr); + } } return (pReader->code != TSDB_CODE_SUCCESS) ? pReader->code : code; @@ -4109,6 +4097,11 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) { tsdbDataFileReaderClose(&pReader->pFileReader); + int64_t loadBlocks = 0; + double elapse = 0; + pReader->status.pLDataIterArray = destroySttBlockReader(pReader->status.pLDataIterArray, &loadBlocks, &elapse); + pReader->status.pLDataIterArray = taosArrayInit(4, POINTER_BYTES); + // resetDataBlockScanInfo excluding lastKey STableBlockScanInfo** p = NULL; int32_t iter = 0; @@ -4179,7 +4172,7 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) { } } - tsdbUntakeReadSnap(pReader, pReader->pReadSnap, false); + tsdbUntakeReadSnap2(pReader, pReader->pReadSnap, false); pReader->pReadSnap = NULL; pReader->flag = READER_STATUS_SUSPEND; diff --git a/source/dnode/vnode/src/tsdb/tsdbRetention.c b/source/dnode/vnode/src/tsdb/tsdbRetention.c index ae5ac4ae36..3af9d2a07a 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRetention.c +++ b/source/dnode/vnode/src/tsdb/tsdbRetention.c @@ -125,6 +125,7 @@ _exit: typedef struct { STsdb *tsdb; + int32_t sync; int64_t now; } SRtnArg; @@ -251,28 +252,33 @@ _exit: if (code) { TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code); } - taosMemoryFree(arg); return code; } -int32_t tsdbAsyncRetention(STsdb *tsdb, int64_t now, int64_t *taskid) { +static void tsdbFreeRtnArg(void *arg) { + SRtnArg *rArg = (SRtnArg *)arg; + if (rArg->sync) { + tsem_post(&rArg->tsdb->pVnode->canCommit); + } + taosMemoryFree(arg); +} + +int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync) { SRtnArg *arg = taosMemoryMalloc(sizeof(*arg)); if (arg == NULL) return TSDB_CODE_OUT_OF_MEMORY; - arg->tsdb = tsdb; + arg->sync = sync; arg->now = now; - int32_t code = tsdbFSScheduleBgTask(tsdb->pFS, TSDB_BG_TASK_RETENTION, tsdbDoRetention2, arg, taskid); - if (code) taosMemoryFree(arg); + if (sync) { + tsem_wait(&tsdb->pVnode->canCommit); + } - return code; -} - -int32_t tsdbSyncRetention(STsdb *tsdb, int64_t now) { int64_t taskid; - - int32_t code = tsdbAsyncRetention(tsdb, now, &taskid); - if (code) return code; - - return tsdbFSWaitBgTask(tsdb->pFS, taskid); + int32_t code = + tsdbFSScheduleBgTask(tsdb->pFS, TSDB_BG_TASK_RETENTION, tsdbDoRetention2, tsdbFreeRtnArg, arg, &taskid); + if (code) { + tsdbFreeRtnArg(arg); + } + return code; } \ No newline at end of file diff --git a/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c b/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c index db8d14d228..27fae9dc6e 100644 --- a/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c +++ b/source/dnode/vnode/src/tsdb/tsdbSttFileRW.c @@ -975,6 +975,11 @@ int32_t tsdbSttFileWriteTombRecord(SSttFileWriter *writer, const STombRecord *re _exit: if (code) { TSDB_ERROR_LOG(TD_VID(writer->config->tsdb->pVnode), lino, code); + } else { + tsdbTrace("vgId:%d write tomb record to stt file:%s, cid:%" PRId64 ", suid:%" PRId64 ", uid:%" PRId64 + ", version:%" PRId64, + TD_VID(writer->config->tsdb->pVnode), writer->fd->path, writer->config->cid, record->suid, record->uid, + record->version); } return code; } diff --git a/source/dnode/vnode/src/vnd/vnodeInitApi.c b/source/dnode/vnode/src/vnd/vnodeInitApi.c index 3dfaa28c09..5c8d563d73 100644 --- a/source/dnode/vnode/src/vnd/vnodeInitApi.c +++ b/source/dnode/vnode/src/vnd/vnodeInitApi.c @@ -180,6 +180,8 @@ void initStateStoreAPI(SStateStore* pStore) { pStore->updateInfoIsUpdated = updateInfoIsUpdated; pStore->updateInfoIsTableInserted = updateInfoIsTableInserted; pStore->updateInfoDestroy = updateInfoDestroy; + pStore->windowSBfDelete = windowSBfDelete; + pStore->windowSBfAdd = windowSBfAdd; pStore->updateInfoInitP = updateInfoInitP; pStore->updateInfoAddCloseWindowSBF = updateInfoAddCloseWindowSBF; diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c index 3b07ef9ea9..65fc552365 100644 --- a/source/dnode/vnode/src/vnd/vnodeOpen.c +++ b/source/dnode/vnode/src/vnd/vnodeOpen.c @@ -476,8 +476,8 @@ void vnodeClose(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); vnodeSyncClose(pVnode); vnodeQueryClose(pVnode); - walClose(pVnode->pWal); tqClose(pVnode->pTq); + walClose(pVnode->pWal); if (pVnode->pTsdb) tsdbClose(&pVnode->pTsdb); smaClose(pVnode->pSma); if (pVnode->pMeta) metaClose(&pVnode->pMeta); diff --git a/source/dnode/vnode/src/vnd/vnodeRetention.c b/source/dnode/vnode/src/vnd/vnodeRetention.c index 7af1f8e28f..f3344d1d7d 100644 --- a/source/dnode/vnode/src/vnd/vnodeRetention.c +++ b/source/dnode/vnode/src/vnd/vnodeRetention.c @@ -15,27 +15,8 @@ #include "vnd.h" -extern int32_t tsdbSyncRetention(STsdb *tsdb, int64_t now); -extern int32_t tsdbAsyncRetention(STsdb *tsdb, int64_t now, int64_t *taskid); +extern int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync); int32_t vnodeDoRetention(SVnode *pVnode, int64_t now) { - int32_t code; - int32_t lino; - - if (pVnode->config.sttTrigger == 1) { - tsem_wait(&pVnode->canCommit); - code = tsdbSyncRetention(pVnode->pTsdb, now); - TSDB_CHECK_CODE(code, lino, _exit); - - // code = smaDoRetention(pVnode->pSma, now); - // TSDB_CHECK_CODE(code, lino, _exit); - tsem_post(&pVnode->canCommit); - } else { - int64_t taskid; - code = tsdbAsyncRetention(pVnode->pTsdb, now, &taskid); - TSDB_CHECK_CODE(code, lino, _exit); - } - -_exit: - return code; + return tsdbRetention(pVnode->pTsdb, now, pVnode->config.sttTrigger == 1); } \ No newline at end of file diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 738809f588..da4bd1e23c 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -2424,7 +2424,9 @@ void streamScanReloadState(SOperatorInfo* pOperator) { pInfo->stateStore.updateInfoDestroy(pInfo->pUpdateInfo); pInfo->pUpdateInfo = pUpInfo; } else { - pInfo->pUpdateInfo->minTS = TMAX(pInfo->pUpdateInfo->minTS, pUpInfo->minTS); + pInfo->stateStore.windowSBfDelete(pInfo->pUpdateInfo, 1); + pInfo->stateStore.windowSBfAdd(pInfo->pUpdateInfo, 1); + ASSERT(pInfo->pUpdateInfo->minTS > pUpInfo->minTS); pInfo->pUpdateInfo->maxDataVersion = TMAX(pInfo->pUpdateInfo->maxDataVersion, pUpInfo->maxDataVersion); SHashObj* curMap = pInfo->pUpdateInfo->pMap; void *pIte = taosHashIterate(curMap, NULL); diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 93a84c6f04..3a5ff91f68 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -4057,13 +4057,20 @@ bool isEqualStateKey(SStateWindowInfo* pWin, char* pKeyData) { bool compareStateKey(void* data, void* key) { if (!data || !key) { - return false; + return true; } SStateKeys* stateKey = (SStateKeys*)key; stateKey->pData = (char*)key + sizeof(SStateKeys); return compareVal(data, stateKey); } +bool compareWinStateKey(SStateKeys* left, SStateKeys* right) { + if (!left || !right) { + return false; + } + return compareVal(left->pData, right); +} + void setStateOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, char* pKeyData, SStateWindowInfo* pCurWin, SStateWindowInfo* pNextWin) { int32_t size = pAggSup->resultRowSize; @@ -4086,10 +4093,14 @@ void setStateOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, pCurWin->winInfo.pOutputBuf = taosMemoryCalloc(1, size); pCurWin->pStateKey = (SStateKeys*)((char*)pCurWin->winInfo.pOutputBuf + (pAggSup->resultRowSize - pAggSup->stateKeySize)); - pCurWin->pStateKey->bytes = pAggSup->stateKeySize - sizeof(SStateKeys); - pCurWin->pStateKey->type = pAggSup->stateKeyType; - pCurWin->pStateKey->pData = (char*)pCurWin->pStateKey + sizeof(SStateKeys); - pCurWin->pStateKey->isNull = false; + pCurWin->pStateKey->bytes = pAggSup->stateKeySize - sizeof(SStateKeys); + pCurWin->pStateKey->type = pAggSup->stateKeyType; + pCurWin->pStateKey->pData = (char*)pCurWin->pStateKey + sizeof(SStateKeys); + pCurWin->pStateKey->isNull = false; + pCurWin->winInfo.sessionWin.groupId = groupId; + pCurWin->winInfo.sessionWin.win.skey = ts; + pCurWin->winInfo.sessionWin.win.ekey = ts; + qDebug("===stream===reset state win key. skey:%" PRId64 ", endkey:%" PRId64, pCurWin->winInfo.sessionWin.win.skey, pCurWin->winInfo.sessionWin.win.ekey); } if (code == TSDB_CODE_SUCCESS) { @@ -4241,6 +4252,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) { SExprSupp* pSup = &pOperator->exprSupp; SStreamStateAggOperatorInfo* pInfo = pOperator->info; SOptrBasicInfo* pBInfo = &pInfo->binfo; + qDebug("===stream=== stream state agg"); if (pOperator->status == OP_RES_TO_RETURN) { doBuildDeleteDataBlock(pOperator, pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator); if (pInfo->pDelRes->info.rows > 0) { @@ -4340,6 +4352,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) { void streamStateReleaseState(SOperatorInfo* pOperator) { SStreamStateAggOperatorInfo* pInfo = pOperator->info; int32_t resSize = taosArrayGetSize(pInfo->historyWins) * sizeof(SSessionKey); + qDebug("===stream=== relase state. save result count:%d", (int32_t)taosArrayGetSize(pInfo->historyWins)); pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_STATE_OP_STATE_NAME, strlen(STREAM_STATE_OP_STATE_NAME), pInfo->historyWins->pData, resSize); SOperatorInfo* downstream = pOperator->pDownstream[0]; if (downstream->fpSet.releaseStreamStateFn) { @@ -4365,6 +4378,7 @@ static void compactStateWindow(SOperatorInfo* pOperator, SResultWindowInfo* pCur compactFunctions(pSup->pCtx, pAggSup->pDummyCtx, numOfOutput, pTaskInfo, &pInfo->twAggSup.timeWindowData); tSimpleHashRemove(pStUpdated, &pNextWin->sessionWin, sizeof(SSessionKey)); if (pNextWin->isOutput && pStDeleted) { + qDebug("===stream=== save delete window info %" PRId64 ", %" PRIu64, pNextWin->sessionWin.win.skey, pNextWin->sessionWin.groupId); saveDeleteRes(pStDeleted, pNextWin->sessionWin); } removeSessionResult(pStUpdated, pAggSup->pResultRows, pNextWin->sessionWin); @@ -4383,19 +4397,28 @@ void streamStateReloadState(SOperatorInfo* pOperator) { int32_t code = pAggSup->stateStore.streamStateGetInfo(pAggSup->pState, STREAM_STATE_OP_STATE_NAME, strlen(STREAM_STATE_OP_STATE_NAME), &pBuf, &size); int32_t num = size / sizeof(SSessionKey); + qDebug("===stream=== reload state. get result count:%d", num); SSessionKey* pSeKeyBuf = (SSessionKey*) pBuf; ASSERT(size == num * sizeof(SSessionKey)); if (!pInfo->pSeUpdated && num > 0) { _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); pInfo->pSeUpdated = tSimpleHashInit(64, hashFn); } + if (!pInfo->pSeDeleted && num > 0) { + _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY); + pInfo->pSeDeleted = tSimpleHashInit(64, hashFn); + } for (int32_t i = 0; i < num; i++) { SStateWindowInfo curInfo = {0}; SStateWindowInfo nextInfo = {0}; SStateWindowInfo dummy = {0}; + qDebug("===stream=== reload state. try process result %" PRId64 ", %" PRIu64 ", index:%d", pSeKeyBuf[i].win.skey, pSeKeyBuf[i].groupId, i); setStateOutputBuf(pAggSup, pSeKeyBuf[i].win.skey, pSeKeyBuf[i].groupId, NULL, &curInfo, &nextInfo); - if (compareStateKey(curInfo.pStateKey,nextInfo.pStateKey)) { - compactStateWindow(pOperator, &curInfo.winInfo, &nextInfo.winInfo, pInfo->pSeUpdated, pInfo->pSeUpdated); + bool cpRes = compareWinStateKey(curInfo.pStateKey,nextInfo.pStateKey); + qDebug("===stream=== reload state. next window info %" PRId64 ", %" PRIu64 ", compare:%d", nextInfo.winInfo.sessionWin.win.skey, nextInfo.winInfo.sessionWin.groupId, cpRes); + if (cpRes) { + compactStateWindow(pOperator, &curInfo.winInfo, &nextInfo.winInfo, pInfo->pSeUpdated, pInfo->pSeDeleted); + qDebug("===stream=== reload state. save result %" PRId64 ", %" PRIu64, curInfo.winInfo.sessionWin.win.skey, curInfo.winInfo.sessionWin.groupId); if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) { saveResult(curInfo.winInfo, pInfo->pSeUpdated); } else if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) { @@ -4406,15 +4429,13 @@ void streamStateReloadState(SOperatorInfo* pOperator) { getSessionHashKey(&curInfo.winInfo.sessionWin, &key); tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &curInfo.winInfo, sizeof(SResultWindowInfo)); } + } else if (IS_VALID_SESSION_WIN(nextInfo.winInfo)) { + releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)nextInfo.winInfo.pOutputBuf, &pAggSup->pSessionAPI->stateStore); } if (IS_VALID_SESSION_WIN(curInfo.winInfo)) { saveSessionOutputBuf(pAggSup, &curInfo.winInfo); } - - if (IS_VALID_SESSION_WIN(nextInfo.winInfo)) { - releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)nextInfo.winInfo.pOutputBuf, &pAggSup->pSessionAPI->stateStore); - } } taosMemoryFree(pBuf); diff --git a/source/libs/parser/inc/parUtil.h b/source/libs/parser/inc/parUtil.h index 1a4ee3e91a..d96bb9bba4 100644 --- a/source/libs/parser/inc/parUtil.h +++ b/source/libs/parser/inc/parUtil.h @@ -118,6 +118,12 @@ int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes); void destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request); SNode* createSelectStmtImpl(bool isDistinct, SNodeList* pProjectionList, SNode* pTable); +/** + * @brief return a - b with overflow check + * @retval val range between [INT64_MIN, INT64_MAX] + */ +int64_t int64SafeSub(int64_t a, int64_t b); + #ifdef __cplusplus } #endif diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index ef1bc8788c..8ce68a5c8c 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -3296,23 +3296,25 @@ static int32_t checkFill(STranslateContext* pCxt, SFillNode* pFill, SValueNode* if (NULL == pInterval) { return TSDB_CODE_SUCCESS; } - - int64_t timeRange = TABS(pFill->timeRange.skey - pFill->timeRange.ekey); + int64_t timeRange = 0; int64_t intervalRange = 0; - if (IS_CALENDAR_TIME_DURATION(pInterval->unit)) { - int64_t f = 1; - if (pInterval->unit == 'n') { - f = 30LL * MILLISECOND_PER_DAY; - } else if (pInterval->unit == 'y') { - f = 365LL * MILLISECOND_PER_DAY; + if (!pCxt->createStream) { + int64_t res = int64SafeSub(pFill->timeRange.skey, pFill->timeRange.ekey); + timeRange = res < 0 ? res == INT64_MIN ? INT64_MAX : -res : res; + if (IS_CALENDAR_TIME_DURATION(pInterval->unit)) { + int64_t f = 1; + if (pInterval->unit == 'n') { + f = 30LL * MILLISECOND_PER_DAY; + } else if (pInterval->unit == 'y') { + f = 365LL * MILLISECOND_PER_DAY; + } + intervalRange = pInterval->datum.i * f; + } else { + intervalRange = pInterval->datum.i; + } + if ((timeRange / intervalRange) >= MAX_INTERVAL_TIME_WINDOW) { + return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_FILL_TIME_RANGE); } - intervalRange = pInterval->datum.i * f; - } else { - intervalRange = pInterval->datum.i; - } - - if ((timeRange / intervalRange) >= MAX_INTERVAL_TIME_WINDOW) { - return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_FILL_TIME_RANGE); } return TSDB_CODE_SUCCESS; diff --git a/source/libs/parser/src/parUtil.c b/source/libs/parser/src/parUtil.c index 42a0d1282a..1c292b1ec4 100644 --- a/source/libs/parser/src/parUtil.c +++ b/source/libs/parser/src/parUtil.c @@ -1142,3 +1142,18 @@ void destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request) { taosHashCleanup(pMetaCache->pTableIndex); taosHashCleanup(pMetaCache->pTableCfg); } + +int64_t int64SafeSub(int64_t a, int64_t b) { + int64_t res = (uint64_t)a - (uint64_t)b; + + if (a >= 0 && b < 0) { + if ((uint64_t)res > (uint64_t)INT64_MAX) { + // overflow + res = INT64_MAX; + } + } else if (a < 0 && b > 0 && res >= 0) { + // underflow + res = INT64_MIN; + } + return res; +} diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index bbd26d0442..34370ebce9 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -581,6 +581,7 @@ int32_t streamTryExec(SStreamTask* pTask) { if (code != TSDB_CODE_SUCCESS) { return code; } + streamSchedExec(pTask); } else { atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE); qDebug("s-task:%s exec completed, status:%s, sched-status:%d", id, streamGetTaskStatusStr(pTask->status.taskStatus), diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c index 0a4f73a67c..5b42be182c 100644 --- a/source/libs/stream/src/streamState.c +++ b/source/libs/stream/src/streamState.c @@ -729,6 +729,7 @@ void streamStateFreeVal(void* val) { int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen) { #ifdef USE_ROCKSDB + qDebug("===stream===save skey:%" PRId64 ", ekey:%" PRId64 ", groupId:%" PRIu64, key->win.skey,key->win.ekey, key->groupId); return streamStateSessionPut_rocksdb(pState, key, value, vLen); #else SStateSessionKey sKey = {.key = *key, .opNum = pState->number}; diff --git a/source/libs/stream/src/streamUpdate.c b/source/libs/stream/src/streamUpdate.c index 85be120dbd..7a8de91d77 100644 --- a/source/libs/stream/src/streamUpdate.c +++ b/source/libs/stream/src/streamUpdate.c @@ -33,7 +33,7 @@ static int64_t adjustExpEntries(int64_t entries) { return TMIN(DEFAULT_EXPECTED_ENTRIES, entries); } -static void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) { +void windowSBfAdd(SUpdateInfo *pInfo, uint64_t count) { if (pInfo->numSBFs < count) { count = pInfo->numSBFs; } @@ -49,7 +49,7 @@ static void clearItemHelper(void *p) { tScalableBfDestroy(*pBf); } -static void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count) { +void windowSBfDelete(SUpdateInfo *pInfo, uint64_t count) { if (count < pInfo->numSBFs) { for (uint64_t i = 0; i < count; ++i) { SScalableBf *pTsSBFs = taosArrayGetP(pInfo->pTsSBFs, 0); diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c index 7ff7fe748e..038fbe444b 100644 --- a/source/libs/wal/src/walRead.c +++ b/source/libs/wal/src/walRead.c @@ -371,9 +371,9 @@ int32_t walFetchHead(SWalReader *pRead, int64_t ver, SWalCkHead *pHead) { pRead->pWal->vers.appliedVer); // TODO: valid ver -// if (ver > pRead->pWal->vers.appliedVer) { -// return -1; -// } + if (ver > pRead->pWal->vers.commitVer) { + return -1; + } if (pRead->curVersion != ver) { code = walReaderSeekVer(pRead, ver); diff --git a/source/util/CMakeLists.txt b/source/util/CMakeLists.txt index 6c9aff046c..9c2ed190c1 100644 --- a/source/util/CMakeLists.txt +++ b/source/util/CMakeLists.txt @@ -5,6 +5,13 @@ if (DEFINED GRANT_CFG_INCLUDE_DIR) add_definitions(-DGRANTS_CFG) endif() +IF (${ASSERT_NOT_CORE}) + ADD_DEFINITIONS(-DASSERT_NOT_CORE) + MESSAGE(STATUS "disable assert core") +ELSE () + MESSAGE(STATUS "enable assert core") +ENDIF (${ASSERT_NOT_CORE}) + target_include_directories( util PUBLIC "${TD_SOURCE_DIR}/include/util" diff --git a/source/util/src/terror.c b/source/util/src/terror.c index e330a659d5..b0b407e2a5 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -643,6 +643,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_CONSUMER_CLOSED, "Consumer closed") TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_CONSUMER_ERROR, "Consumer error, to see log") TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE, "Topic num out of range") TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_GROUP_OUT_OF_RANGE, "Group num out of range 100") +TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_SAME_COMMITTED_VALUE, "Same committed value") // stream TAOS_DEFINE_ERROR(TSDB_CODE_STREAM_TASK_NOT_EXIST, "Stream task not exist") diff --git a/source/util/src/tlog.c b/source/util/src/tlog.c index c07bafa1ea..de7ad848ed 100644 --- a/source/util/src/tlog.c +++ b/source/util/src/tlog.c @@ -76,7 +76,11 @@ static int32_t tsDaylightActive; /* Currently in daylight saving time. */ bool tsLogEmbedded = 0; bool tsAsyncLog = true; +#ifdef ASSERT_NOT_CORE +bool tsAssert = false; +#else bool tsAssert = true; +#endif int32_t tsNumOfLogLines = 10000000; int32_t tsLogKeepDays = 0; LogFp tsLogFp = NULL; diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 1ec5102d9b..89572d1c06 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -105,7 +105,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb-mutilVg-mutilCtb-funcNFilter.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb-mutilVg-mutilCtb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-1ctb-funcNFilter.py -#,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py +,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqAutoCreateTbl.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDnodeRestart.py @@ -345,6 +345,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/smaTest.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/smaTest.py -R ,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/sma_index.py +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml_TS-3724.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -R ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/spread.py diff --git a/tests/script/sh/stop_dnodes.bat b/tests/script/sh/stop_dnodes.bat index ab7af2ca92..65aee26ed4 100644 --- a/tests/script/sh/stop_dnodes.bat +++ b/tests/script/sh/stop_dnodes.bat @@ -3,4 +3,6 @@ rem echo taskkill /F /IM taosd.exe wmic process where "name='taosd.exe'" call terminate > NUL 2>&1 -taskkill /F /IM taosd.exe > NUL 2>&1 \ No newline at end of file +taskkill /F /IM taosd.exe > NUL 2>&1 + +rem echo taskkill /F /IM taosd.exe finished \ No newline at end of file diff --git a/tests/system-test/1-insert/opentsdb_json_taosc_insert.py b/tests/system-test/1-insert/opentsdb_json_taosc_insert.py index 857a8e3a32..91ef1e70a3 100644 --- a/tests/system-test/1-insert/opentsdb_json_taosc_insert.py +++ b/tests/system-test/1-insert/opentsdb_json_taosc_insert.py @@ -24,6 +24,8 @@ import threading import json class TDTestCase: + updatecfgDict = {'clientCfg': {'smlDot2Underline': 0}} + def init(self, conn, logSql, replicaVar=1): self.replicaVar = int(replicaVar) tdLog.debug("start to execute %s" % __file__) diff --git a/tests/system-test/1-insert/opentsdb_telnet_line_taosc_insert.py b/tests/system-test/1-insert/opentsdb_telnet_line_taosc_insert.py index 351cf49e3a..d419aee12c 100644 --- a/tests/system-test/1-insert/opentsdb_telnet_line_taosc_insert.py +++ b/tests/system-test/1-insert/opentsdb_telnet_line_taosc_insert.py @@ -28,6 +28,8 @@ if platform.system().lower() == 'windows': sys.stdout = io.TextIOWrapper(sys.stdout.buffer,encoding='utf8') class TDTestCase: + updatecfgDict = {'clientCfg': {'smlDot2Underline': 0}} + def init(self, conn, logSql, replicaVar=1): self.replicaVar = int(replicaVar) tdLog.debug("start to execute %s" % __file__) diff --git a/tests/system-test/2-query/sml.py b/tests/system-test/2-query/sml.py index 2f97118fbf..b3aeb72194 100644 --- a/tests/system-test/2-query/sml.py +++ b/tests/system-test/2-query/sml.py @@ -15,7 +15,7 @@ sys.path.append("./7-tmq") from tmqCommon import * class TDTestCase: - updatecfgDict = {'clientCfg': {'smlChildTableName': 'dataModelName', 'fqdn': 'localhost'}, 'fqdn': 'localhost'} + updatecfgDict = {'clientCfg': {'smlChildTableName': 'dataModelName', 'fqdn': 'localhost', 'smlDot2Underline': 0}, 'fqdn': 'localhost'} print("===================: ", updatecfgDict) def init(self, conn, logSql, replicaVar=1): @@ -101,6 +101,15 @@ class TDTestCase: tdSql.query(f"desc {dbname}.macylr") tdSql.checkRows(25) + + tdSql.query(f"select * from ts3724.`.stb2`") + tdSql.checkRows(1) + + tdSql.query(f"select * from ts3724.`stb.2`") + tdSql.checkRows(1) + + tdSql.query(f"select * from ts3724.`stb2.`") + tdSql.checkRows(1) return def run(self): diff --git a/tests/system-test/2-query/sml_TS-3724.py b/tests/system-test/2-query/sml_TS-3724.py new file mode 100644 index 0000000000..a8b16c4662 --- /dev/null +++ b/tests/system-test/2-query/sml_TS-3724.py @@ -0,0 +1,125 @@ +import taos +import sys +import time +import socket +import os +import threading + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +sys.path.append("./7-tmq") +from tmqCommon import * + +class TDTestCase: + updatecfgDict = {'clientCfg': {'smlChildTableName': 'dataModelName', 'fqdn': 'localhost', 'smlTsDefaultName': "times"}, 'fqdn': 'localhost'} + print("===================: ", updatecfgDict) + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor(), True) + #tdSql.init(conn.cursor(), logSql) # output sql.txt file + + def checkContent(self, dbname="sml_db"): + simClientCfg="%s/taos.cfg"%tdDnodes.getSimCfgPath() + buildPath = tdCom.getBuildPath() + cmdStr = '%s/build/bin/sml_test %s'%(buildPath, simClientCfg) + print("cmdStr:", cmdStr) + tdLog.info(cmdStr) + ret = os.system(cmdStr) + if ret != 0: + tdLog.info("sml_test ret != 0") + + tdSql.query(f"select * from ts3303.stb2") + tdSql.query(f"select * from ts3303.meters") + + # tdSql.execute('use sml_db') + tdSql.query(f"select * from {dbname}.t_b7d815c9222ca64cdf2614c61de8f211") + tdSql.checkRows(1) + + tdSql.checkData(0, 0, '2016-01-01 08:00:07.000') + tdSql.checkData(0, 1, 2000) + tdSql.checkData(0, 2, 200) + tdSql.checkData(0, 3, 15) + tdSql.checkData(0, 4, 24.5208) + tdSql.checkData(0, 5, 28.09377) + tdSql.checkData(0, 6, 428) + tdSql.checkData(0, 7, 0) + tdSql.checkData(0, 8, 304) + tdSql.checkData(0, 9, 0) + tdSql.checkData(0, 10, 25) + + tdSql.query(f"select * from {dbname}.readings") + tdSql.checkRows(9) + + tdSql.query(f"select distinct tbname from {dbname}.readings") + tdSql.checkRows(4) + + tdSql.query(f"select * from {dbname}.t_0799064f5487946e5d22164a822acfc8 order by times") + tdSql.checkRows(2) + tdSql.checkData(0, 3, "kk") + tdSql.checkData(1, 3, "") + + + tdSql.query(f"select distinct tbname from {dbname}.`sys_if_bytes_out`") + tdSql.checkRows(2) + + tdSql.query(f"select * from {dbname}.t_fc70dec6677d4277c5d9799c4da806da order by times") + tdSql.checkRows(2) + tdSql.checkData(0, 1, 1.300000000) + tdSql.checkData(1, 1, 13.000000000) + + tdSql.query(f"select * from {dbname}.`sys_procs_running`") + tdSql.checkRows(1) + tdSql.checkData(0, 1, 42.000000000) + tdSql.checkData(0, 2, "web01") + + tdSql.query(f"select distinct tbname from {dbname}.`sys_cpu_nice`") + tdSql.checkRows(3) + + tdSql.query(f"select * from {dbname}.`sys_cpu_nice` order by times") + tdSql.checkRows(4) + tdSql.checkData(0, 1, 13.000000000) + tdSql.checkData(0, 2, "web01") + tdSql.checkData(0, 3, None) + tdSql.checkData(0, 4, "lga") + + tdSql.checkData(1, 1, 9.000000000) + tdSql.checkData(1, 2, "web02") + tdSql.checkData(3, 3, "t1") + tdSql.checkData(0, 4, "lga") + + tdSql.query(f"select * from {dbname}.macylr") + tdSql.checkRows(2) + + tdSql.query(f"select * from {dbname}.qelhxo") + tdSql.checkRows(5) + + tdSql.query(f"desc {dbname}.macylr") + tdSql.checkRows(25) + + tdSql.query(f"select * from ts3724._stb2") + tdSql.checkRows(1) + + tdSql.query(f"select * from ts3724.stb_2") + tdSql.checkRows(1) + + tdSql.query(f"select * from ts3724.stb2_") + tdSql.checkRows(1) + return + + def run(self): + tdSql.prepare() + self.checkContent() + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) + diff --git a/tests/system-test/7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py b/tests/system-test/7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py index 37946d0c22..a717c4966d 100644 --- a/tests/system-test/7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py +++ b/tests/system-test/7-tmq/tmqConsFromTsdb1-mutilVg-mutilCtb-funcNFilter.py @@ -222,9 +222,9 @@ class TDTestCase: actConsumeTotalRows = resultList[0] - if not (actConsumeTotalRows > 0 and actConsumeTotalRows < totalRowsInserted): + if not (actConsumeTotalRows >= 0 and actConsumeTotalRows <= totalRowsInserted): tdLog.info("act consume rows: %d"%(actConsumeTotalRows)) - tdLog.info("and second consume rows should be between 0 and %d"%(totalRowsInserted)) + tdLog.info("and second consume rows should be between [0 and %d]"%(totalRowsInserted)) tdLog.exit("%d tmq consume rows error!"%consumerId) time.sleep(10) diff --git a/tools/shell/src/shellUtil.c b/tools/shell/src/shellUtil.c index e15b49efcc..93451c85a9 100644 --- a/tools/shell/src/shellUtil.c +++ b/tools/shell/src/shellUtil.c @@ -68,7 +68,7 @@ int32_t shellCheckIntSize() { return 0; } -void shellPrintVersion() { printf("version: %s\r\n", version); } +void shellPrintVersion() { printf("%s\r\n", shell.info.programVersion); } void shellGenerateAuth() { char secretEncrypt[TSDB_PASSWORD_LEN + 1] = {0}; diff --git a/utils/test/c/sml_test.c b/utils/test/c/sml_test.c index dad30db02a..e4ed6037a3 100644 --- a/utils/test/c/sml_test.c +++ b/utils/test/c/sml_test.c @@ -1522,6 +1522,36 @@ int sml_ts2385_Test() { return code; } +int sml_ts3724_Test() { + TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0); + + TAOS_RES *pRes = taos_query(taos, "drop database if exists ts3724"); + taos_free_result(pRes); + + pRes = taos_query(taos, "create database if not exists ts3724"); + taos_free_result(pRes); + + const char *sql[] = { + "stb.2,t1=1 f1=283i32 1632299372000", + ".stb2,t1=1 f1=106i32 1632299378000", + "stb2.,t1=1 f1=106i32 1632299378000", + }; + + pRes = taos_query(taos, "use ts3724"); + taos_free_result(pRes); + + pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_LINE_PROTOCOL, + TSDB_SML_TIMESTAMP_MILLI_SECONDS); + + int code = taos_errno(pRes); + printf("%s result0:%s\n", __FUNCTION__, taos_errstr(pRes)); + taos_free_result(pRes); + + taos_close(taos); + + return code; +} + int main(int argc, char *argv[]) { if (argc == 2) { taos_options(TSDB_OPTION_CONFIGDIR, argv[1]); @@ -1579,5 +1609,8 @@ int main(int argc, char *argv[]) { ASSERT(!ret); ret = sml_19221_Test(); ASSERT(!ret); + ret = sml_ts3724_Test(); + ASSERT(!ret); + return ret; } diff --git a/utils/tsim/CMakeLists.txt b/utils/tsim/CMakeLists.txt index 81737809d9..c2cf7ac3c5 100644 --- a/utils/tsim/CMakeLists.txt +++ b/utils/tsim/CMakeLists.txt @@ -2,7 +2,7 @@ aux_source_directory(src TSIM_SRC) add_executable(tsim ${TSIM_SRC}) target_link_libraries( tsim - PUBLIC taos + PUBLIC taos_static PUBLIC util PUBLIC common PUBLIC os