Merge branch 'main' of https://github.com/taosdata/TDengine into mark/tmq

docs/en/12-taos-sql/25-grant.md

@@ -16,7 +16,7 @@ This statement creates a user account.
 
 The maximum length of user_name is 23 bytes.
 
-The maximum length of password is 128 bytes. The password can include leters, digits, and special characters excluding single quotation marks, double quotation marks, backticks, backslashes, and spaces. The password cannot be empty.
+The maximum length of password is 32 bytes. The password can include leters, digits, and special characters excluding single quotation marks, double quotation marks, backticks, backslashes, and spaces. The password cannot be empty.
 
 `SYSINFO` indicates whether the user is allowed to view system information. `1` means allowed, `0` means not allowed. System information includes server configuration, dnode, vnode, storage. The default value is `1`.
 

docs/en/14-reference/12-config/index.md

@@ -102,7 +102,7 @@ Ensure that your firewall rules do not block TCP port 6042  on any host in the c
 | Value Range   | 10-50000000                                          |
 | Default Value | 5000                                                 |
 
-### numOfRpcSessions 
+### numOfRpcSessions
 
 | Attribute     | Description                                |
 | ------------- | ------------------------------------------ |
@@ -202,7 +202,7 @@ Please note the `taoskeeper` needs to be installed and running to create the `lo
 | Default Value | 0                                                                                                                                                                   |
 | Notes         | 0: Disable SMA indexing and perform all queries on non-indexed data; 1: Enable SMA indexing and perform queries from suitable statements on precomputation results. |
 
-### countAlwaysReturnValue 
+### countAlwaysReturnValue
 
 | Attribute  | Description                                                                                                                                                                                                                     |
 | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
@@ -713,6 +713,14 @@ The charset that takes effect is UTF-8.
 | Value Range   | 0: disable UDF; 1: enabled UDF     |
 | Default Value | 1                                  |
 
+### ttlChangeOnWrite
+
+| Attribute     | Description                                                                   |
+| ------------- | ----------------------------------------------------------------------------- |
+| Applicable    | Server Only                                                                   |
+| Meaning       | Whether the ttl expiration time changes with the table modification operation |
+| Value Range   | 0: not change; 1: change by modification                                      |
+| Default Value | 0                                                                             |
 
 ## 3.0 Parameters
 
@@ -770,3 +778,4 @@ The charset that takes effect is UTF-8.
 | 52  |        charset         | Yes                    | Yes                          |                         |
 | 53  |          udf           | Yes                    | Yes                          |                         |
 | 54  |     enableCoreFile     | Yes                    | Yes                          |                         |
+| 55  |    ttlChangeOnWrite    | No                     | Yes                          |                         |

docs/zh/12-taos-sql/25-grant.md

@@ -16,7 +16,7 @@ CREATE USER use_name PASS 'password' [SYSINFO {1|0}];
 
 use_name 最长为 23 字节。
 
-password 最长为 128 字节，合法字符包括"a-zA-Z0-9!?$%^&*()_–+={[}]:;@~#|<,>.?/"，不可以出现单双引号、撇号、反斜杠和空格，且不可以为空。
+password 最长为 32 字节，合法字符包括"a-zA-Z0-9!?$%^&*()_–+={[}]:;@~#|<,>.?/"，不可以出现单双引号、撇号、反斜杠和空格，且不可以为空。
 
 SYSINFO 表示用户是否可以查看系统信息。1 表示可以查看，0 表示不可以查看。系统信息包括服务端配置信息、服务端各种节点信息（如 DNODE、QNODE等）、存储相关的信息等。默认为可以查看系统信息。
 

docs/zh/14-reference/12-config/index.md

@@ -101,7 +101,7 @@ taos -C
 | 取值范围 | 10-50000000             |
 | 缺省值  | 5000                    |
 
-### numOfRpcSessions 
+### numOfRpcSessions
 
 | 属性     | 说明                    |
 | --------| ---------------------- |
@@ -120,7 +120,7 @@ taos -C
 | 缺省值   | 500000               |
 
 
-### numOfRpcSessions 
+### numOfRpcSessions
 
 | 属性     | 说明                         |
 | -------- | ---------------------------- |
@@ -717,6 +717,15 @@ charset 的有效值是 UTF-8。
 | 取值范围 | 0: 不启动；1：启动 |
 | 缺省值   | 1                  |
 
+### ttlChangeOnWrite
+
+| 属性     | 说明               |
+| -------- | ------------------ |
+| 适用范围 | 仅服务端适用       |
+| 含义     | ttl 到期时间是否伴随表的修改操作改变  |
+| 取值范围 | 0: 不改变；1：改变 |
+| 缺省值   | 0                  |
+
 ## 压缩参数
 
 ### compressMsgSize
@@ -784,6 +793,7 @@ charset 的有效值是 UTF-8。
 | 52  |        charset         | 是              | 是                              |                    |
 | 53  |          udf           | 是              | 是                              |                    |
 | 54  |     enableCoreFile     | 是              | 是                              |                    |
+| 55  |    ttlChangeOnWrite    | 否              | 是                              |                    |
 
 ## 2.x->3.0 的废弃参数
 

source/client/src/clientTmq.c

@@ -1426,7 +1426,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
     tDecoderClear(&decoder);
     memcpy(&pRspWrapper->dataRsp, pMsg->pData, sizeof(SMqRspHead));
 
-    char buf[TSDB_OFFSET_LEN];
+    char buf[TSDB_OFFSET_LEN] = {0};
     tFormatOffset(buf, TSDB_OFFSET_LEN, &pRspWrapper->dataRsp.rspOffset);
     tscDebug("consumer:0x%" PRIx64 " recv poll rsp, vgId:%d, req ver:%" PRId64 ", rsp:%s type %d, reqId:0x%" PRIx64,
              tmq->consumerId, vgId, pRspWrapper->dataRsp.reqOffset.version, buf, rspType, requestId);
@@ -1572,7 +1572,7 @@ static bool doUpdateLocalEp(tmq_t* tmq, int32_t epoch, const SMqAskEpRsp* pRsp)
         SMqClientVg* pVgCur = taosArrayGet(pTopicCur->vgs, j);
         makeTopicVgroupKey(vgKey, pTopicCur->topicName, pVgCur->vgId);
 
-        char buf[TSDB_OFFSET_LEN];
+        char buf[TSDB_OFFSET_LEN] = {0};
         tFormatOffset(buf, TSDB_OFFSET_LEN, &pVgCur->offsetInfo.currentOffset);
         tscInfo("consumer:0x%" PRIx64 ", epoch:%d vgId:%d vgKey:%s, offset:%s", tmq->consumerId, epoch, pVgCur->vgId,
                  vgKey, buf);
@@ -1800,7 +1800,7 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p
   sendInfo->msgType = TDMT_VND_TMQ_CONSUME;
 
   int64_t transporterId = 0;
-  char    offsetFormatBuf[TSDB_OFFSET_LEN];
+  char    offsetFormatBuf[TSDB_OFFSET_LEN] = {0};
   tFormatOffset(offsetFormatBuf, tListLen(offsetFormatBuf), &pVg->offsetInfo.currentOffset);
 
   tscDebug("consumer:0x%" PRIx64 " send poll to %s vgId:%d, epoch %d, req:%s, reqId:0x%" PRIx64, pTmq->consumerId,
@@ -1946,7 +1946,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
 
         updateVgInfo(pVg, &pDataRsp->rspOffset, pDataRsp->head.walsver, pDataRsp->head.walever, tmq->consumerId);
 
-        char buf[TSDB_OFFSET_LEN];
+        char buf[TSDB_OFFSET_LEN] = {0};
         tFormatOffset(buf, TSDB_OFFSET_LEN, &pDataRsp->rspOffset);
         if (pDataRsp->blockNum == 0) {
           tscDebug("consumer:0x%" PRIx64 " empty block received, vgId:%d, offset:%s, vg total:%" PRId64
@@ -2043,7 +2043,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
 
           tmq->totalRows += numOfRows;
 
-          char buf[TSDB_OFFSET_LEN];
+          char buf[TSDB_OFFSET_LEN] = {0};
           tFormatOffset(buf, TSDB_OFFSET_LEN, &pVg->offsetInfo.currentOffset);
           tscDebug("consumer:0x%" PRIx64 " process taosx poll rsp, vgId:%d, offset:%s, blocks:%d, rows:%" PRId64
                        ", vg total:%" PRId64 ", total:%" PRId64 ", reqId:0x%" PRIx64,
@@ -2686,7 +2686,7 @@ int32_t tmq_get_topic_assignment(tmq_t* tmq, const char* pTopicName, tmq_topic_a
       sendInfo->msgType = TDMT_VND_TMQ_VG_WALINFO;
 
       int64_t transporterId = 0;
-      char    offsetFormatBuf[TSDB_OFFSET_LEN];
+      char    offsetFormatBuf[TSDB_OFFSET_LEN] = {0};
       tFormatOffset(offsetFormatBuf, tListLen(offsetFormatBuf), &pClientVg->offsetInfo.currentOffset);
 
       tscInfo("consumer:0x%" PRIx64 " %s retrieve wal info vgId:%d, epoch %d, req:%s, reqId:0x%" PRIx64,

source/dnode/mgmt/exe/dmMain.c

@@ -19,6 +19,9 @@
 #include "tconfig.h"
 #include "tglobal.h"
 #include "version.h"
+#ifdef TD_JEMALLOC_ENABLED
+#include "jemalloc/jemalloc.h"
+#endif
 
 #if defined(CUS_NAME) || defined(CUS_PROMPT) || defined(CUS_EMAIL)
 #include "cus_name.h"
@@ -255,6 +258,10 @@ static void taosCleanupArgs() {
 }
 
 int main(int argc, char const *argv[]) {
+#ifdef TD_JEMALLOC_ENABLED
+  bool jeBackgroundThread = true;
+  mallctl("background_thread", NULL, NULL, &jeBackgroundThread, sizeof(bool));
+#endif
   if (!taosCheckSystemIsLittleEnd()) {
     printf("failed to start since on non-little-end machines\n");
     return -1;

source/dnode/mnode/impl/src/mndConsumer.c

@@ -419,6 +419,9 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
     mDebug("heartbeat report offset rows.%s:%s", pConsumer->cgroup, data->topicName);
 
     SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, pConsumer->cgroup, data->topicName);
+    if(pSub == NULL){
+      continue;
+    }
     taosWLockLatch(&pSub->lock);
     SMqConsumerEp *pConsumerEp = taosHashGet(pSub->consumerHash, &consumerId, sizeof(int64_t));
     if(pConsumerEp){

source/dnode/vnode/src/inc/tq.h

@@ -151,7 +151,8 @@ int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxR
 int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision);
 int32_t tqSendDataRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp,
                       int32_t type, int32_t vgId);
-int32_t tqPushDataRsp(STqHandle* pHandle, int32_t vgId);
+//int32_t tqPushDataRsp(STqHandle* pHandle, int32_t vgId);
+int32_t tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId);
 
 // tqMeta
 int32_t tqMetaOpen(STQ* pTq);

source/dnode/vnode/src/tq/tq.c

@@ -232,26 +232,43 @@ static int32_t doSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqData
   return 0;
 }
 
-int32_t tqPushDataRsp(STqHandle* pHandle, int32_t vgId) {
-  SMqDataRsp dataRsp = {0};
-  dataRsp.head.consumerId = pHandle->consumerId;
-  dataRsp.head.epoch = pHandle->epoch;
-  dataRsp.head.mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
-
-  int64_t sver = 0, ever = 0;
-  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
-  tqDoSendDataRsp(&pHandle->msg->info, &dataRsp, pHandle->epoch, pHandle->consumerId, TMQ_MSG_TYPE__POLL_RSP, sver,
-                  ever);
+int32_t tqPushEmptyDataRsp(STqHandle* pHandle, int32_t vgId) {
+  SMqPollReq req = {0};
+  if (tDeserializeSMqPollReq(pHandle->msg->pCont, pHandle->msg->contLen, &req) < 0) {
+    tqError("tDeserializeSMqPollReq %d failed", pHandle->msg->contLen);
+    terrno = TSDB_CODE_INVALID_MSG;
+    return -1;
+  }
 
-  char buf1[TSDB_OFFSET_LEN] = {0};
-  char buf2[TSDB_OFFSET_LEN] = {0};
-  tFormatOffset(buf1, tListLen(buf1), &dataRsp.reqOffset);
-  tFormatOffset(buf2, tListLen(buf2), &dataRsp.rspOffset);
-  tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) push rsp, block num: %d, req:%s, rsp:%s", vgId,
-          dataRsp.head.consumerId, dataRsp.head.epoch, dataRsp.blockNum, buf1, buf2);
+  SMqDataRsp dataRsp = {0};
+  tqInitDataRsp(&dataRsp, &req);
+  dataRsp.blockNum = 0;
+  dataRsp.rspOffset = dataRsp.reqOffset;
+  tqSendDataRsp(pHandle, pHandle->msg, &req, &dataRsp, TMQ_MSG_TYPE__POLL_RSP, vgId);
+  tDeleteMqDataRsp(&dataRsp);
   return 0;
 }
 
+//int32_t tqPushDataRsp(STqHandle* pHandle, int32_t vgId) {
+//  SMqDataRsp dataRsp = {0};
+//  dataRsp.head.consumerId = pHandle->consumerId;
+//  dataRsp.head.epoch = pHandle->epoch;
+//  dataRsp.head.mqMsgType = TMQ_MSG_TYPE__POLL_RSP;
+//
+//  int64_t sver = 0, ever = 0;
+//  walReaderValidVersionRange(pHandle->execHandle.pTqReader->pWalReader, &sver, &ever);
+//  tqDoSendDataRsp(&pHandle->msg->info, &dataRsp, pHandle->epoch, pHandle->consumerId, TMQ_MSG_TYPE__POLL_RSP, sver,
+//                  ever);
+//
+//  char buf1[TSDB_OFFSET_LEN] = {0};
+//  char buf2[TSDB_OFFSET_LEN] = {0};
+//  tFormatOffset(buf1, tListLen(buf1), &dataRsp.reqOffset);
+//  tFormatOffset(buf2, tListLen(buf2), &dataRsp.rspOffset);
+//  tqDebug("vgId:%d, from consumer:0x%" PRIx64 " (epoch %d) push rsp, block num: %d, req:%s, rsp:%s", vgId,
+//          dataRsp.head.consumerId, dataRsp.head.epoch, dataRsp.blockNum, buf1, buf2);
+//  return 0;
+//}
+
 int32_t tqSendDataRsp(STqHandle* pHandle, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp,
                       int32_t type, int32_t vgId) {
   int64_t sver = 0, ever = 0;
@@ -510,7 +527,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
     pHandle->epoch = reqEpoch;
   }
 
-  char buf[TSDB_OFFSET_LEN];
+  char buf[TSDB_OFFSET_LEN] = {0};
   tFormatOffset(buf, TSDB_OFFSET_LEN, &reqOffset);
   tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d), subkey %s, recv poll req vgId:%d, req:%s, reqId:0x%" PRIx64,
           consumerId, req.epoch, pHandle->subKey, vgId, buf, req.reqId);

source/dnode/vnode/src/tq/tqPush.c

@@ -64,7 +64,9 @@ int32_t tqRegisterPushHandle(STQ* pTq, void* handle, SRpcMsg* pMsg) {
     memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));
     pHandle->msg->pCont = rpcMallocCont(pMsg->contLen);
   } else {
-    tqPushDataRsp(pHandle, vgId);
+//    tqPushDataRsp(pHandle, vgId);
+    tqPushEmptyDataRsp(pHandle, vgId);
+
     void* tmp = pHandle->msg->pCont;
     memcpy(pHandle->msg, pMsg, sizeof(SRpcMsg));
     pHandle->msg->pCont = tmp;
@@ -89,7 +91,8 @@ int32_t tqUnregisterPushHandle(STQ* pTq, void *handle) {
   tqDebug("vgId:%d remove pHandle:%p,ret:%d consumer Id:0x%" PRIx64, vgId, pHandle, ret, pHandle->consumerId);
 
   if(pHandle->msg != NULL) {
-    tqPushDataRsp(pHandle, vgId);
+//    tqPushDataRsp(pHandle, vgId);
+    tqPushEmptyDataRsp(pHandle, vgId);
 
     rpcFreeCont(pHandle->msg->pCont);
     taosMemoryFree(pHandle->msg);

source/dnode/vnode/src/tq/tqUtil.c

@@ -99,7 +99,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
   if (pOffset != NULL) {
     *pOffsetVal = pOffset->val;
 
-    char formatBuf[TSDB_OFFSET_LEN];
+    char formatBuf[TSDB_OFFSET_LEN] = {0};
     tFormatOffset(formatBuf, TSDB_OFFSET_LEN, pOffsetVal);
     tqDebug("tmq poll: consumer:0x%" PRIx64
             ", subkey %s, vgId:%d, existed offset found, offset reset to %s and continue. reqId:0x%" PRIx64,
@@ -162,6 +162,7 @@ static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle,
   uint64_t consumerId = pRequest->consumerId;
   int32_t  vgId = TD_VID(pTq->pVnode);
   int      code = 0;
+  terrno        = 0;
 
   SMqDataRsp dataRsp = {0};
   tqInitDataRsp(&dataRsp, pRequest);

source/libs/parser/src/parInsertSml.c

@@ -127,7 +127,7 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
 
     if(kv->keyLen != strlen(pTagSchema->name) || memcmp(kv->key, pTagSchema->name, kv->keyLen) != 0 || kv->type != pTagSchema->type){
       code = TSDB_CODE_SML_INVALID_DATA;
-      uError("SML smlBuildCol error col not same %s", pTagSchema->name);
+      uError("SML smlBuildTagRow error col not same %s", pTagSchema->name);
       goto end;
     }
 
@@ -210,7 +210,7 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
   SSmlKv*  kv = (SSmlKv*)data;
   if(kv->keyLen != strlen(pColSchema->name) || memcmp(kv->key, pColSchema->name, kv->keyLen) != 0 || kv->type != pColSchema->type){
     ret = TSDB_CODE_SML_INVALID_DATA;
-    uError("SML smlBuildCol error col not same %s", pColSchema->name);
+    uInfo("SML smlBuildCol error col not same %s", pColSchema->name);
     goto end;
   }
   if (kv->type == TSDB_DATA_TYPE_NCHAR) {

source/libs/stream/src/streamExec.c

@@ -407,7 +407,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
         break;
       }
 
-      if (status != TASK_STATUS__NORMAL && status != TASK_STATUS__PAUSE) {
+      if (status != TASK_STATUS__NORMAL && status != TASK_STATUS__PAUSE && status != TASK_STATUS__STOP) {
         qError("stream task wait for the end of fill history, s-task:%s, status:%d", id, status);
         taosMsleep(100);
       } else {

source/libs/wal/src/walRead.c

@@ -82,6 +82,11 @@ int32_t walNextValidMsg(SWalReader *pReader) {
          ", applied index:%" PRId64", end index:%" PRId64,
          pReader->pWal->cfg.vgId, fetchVer, lastVer, committedVer, appliedVer, endVer);
 
+  if (fetchVer > endVer){
+    terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
+    return -1;
+  }
+
   while (fetchVer <= endVer) {
     if (walFetchHeadNew(pReader, fetchVer) < 0) {
       return -1;

tests/system-test/7-tmq/subscribeStb3.py

@@ -546,7 +546,7 @@ class TDTestCase:
         keyList        = 'group.id:cgrp1,\
                           enable.auto.commit:false,\
                           auto.commit.interval.ms:6000,\
-                          auto.offset.reset:none'
+                          auto.offset.reset:earliest'
         self.insertConsumerInfo(consumerId, expectrowcnt/2,topicList,keyList,ifcheckdata,ifManualCommit)
 
         tdLog.info("again start consume processor")
@@ -569,7 +569,7 @@ class TDTestCase:
         keyList        = 'group.id:cgrp1,\
                           enable.auto.commit:false,\
                           auto.commit.interval.ms:6000,\
-                          auto.offset.reset:none'
+                          auto.offset.reset:earliest'
         self.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit)
 
         tdLog.info("again start consume processor")