Merge pull request #20367 from taosdata/feat/TS-2846

feat: alter stt trigger and min rows

include/common/tmsg.h

@@ -845,6 +845,7 @@ typedef struct {
   int8_t  cacheLast;
   int8_t  replications;
   int32_t sstTrigger;
+  int32_t minRows;
 } SAlterDbReq;
 
 int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);
@@ -1316,6 +1317,9 @@ typedef struct {
   int8_t  strict;
   int8_t  cacheLast;
   int64_t reserved[8];
+  // 1st modification
+  int16_t sttTrigger;
+  int32_t minRows;
 } SAlterVnodeConfigReq;
 
 int32_t tSerializeSAlterVnodeConfigReq(void* buf, int32_t bufLen, SAlterVnodeConfigReq* pReq);

source/common/src/tmsg.c

@@ -2219,6 +2219,10 @@ int32_t tSerializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
   if (tEncodeI8(&encoder, pReq->cacheLast) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->replications) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->sstTrigger) < 0) return -1;
+
+  // 1st modification
+  if (tEncodeI32(&encoder, pReq->minRows) < 0) return -1;
+
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -2246,6 +2250,13 @@ int32_t tDeserializeSAlterDbReq(void *buf, int32_t bufLen, SAlterDbReq *pReq) {
   if (tDecodeI8(&decoder, &pReq->cacheLast) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->replications) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->sstTrigger) < 0) return -1;
+
+  // 1st modification
+  if (!tDecodeIsEnd(&decoder)) {
+    if (tDecodeI32(&decoder, &pReq->minRows) < 0) return -1;
+  } else {
+    pReq->minRows = -1;
+  }
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);
@@ -4172,6 +4183,11 @@ int32_t tSerializeSAlterVnodeConfigReq(void *buf, int32_t bufLen, SAlterVnodeCon
   for (int32_t i = 0; i < 8; ++i) {
     if (tEncodeI64(&encoder, pReq->reserved[i]) < 0) return -1;
   }
+
+  // 1st modification
+  if (tEncodeI16(&encoder, pReq->sttTrigger) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->minRows) < 0) return -1;
+
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -4201,6 +4217,15 @@ int32_t tDeserializeSAlterVnodeConfigReq(void *buf, int32_t bufLen, SAlterVnodeC
     if (tDecodeI64(&decoder, &pReq->reserved[i]) < 0) return -1;
   }
 
+  // 1st modification
+  if (tDecodeIsEnd(&decoder)) {
+    pReq->sttTrigger = -1;
+    pReq->minRows = -1;
+  } else {
+    if (tDecodeI16(&decoder, &pReq->sttTrigger) < 0) return -1;
+    if (tDecodeI32(&decoder, &pReq->minRows) < 0) return -1;
+  }
+
   tEndDecode(&decoder);
   tDecoderClear(&decoder);
   return 0;

source/dnode/mnode/impl/inc/mndDef.h

@@ -543,7 +543,7 @@ void*           tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer
 
 typedef struct {
   int32_t vgId;
-  char*   qmsg;   //SubPlanToString
+  char*   qmsg;  // SubPlanToString
   SEpSet  epSet;
 } SMqVgEp;
 

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

@@ -725,6 +725,18 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
     terrno = 0;
   }
 
+  if (pAlter->sstTrigger > 0 && pAlter->sstTrigger != pDb->cfg.sstTrigger) {
+    pDb->cfg.sstTrigger = pAlter->sstTrigger;
+    pDb->vgVersion++;
+    terrno = 0;
+  }
+
+  if (pAlter->minRows > 0 && pAlter->minRows != pDb->cfg.minRows) {
+    pDb->cfg.minRows = pAlter->minRows;
+    pDb->vgVersion++;
+    terrno = 0;
+  }
+
   return terrno;
 }
 

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

@@ -319,6 +319,8 @@ static void *mndBuildAlterVnodeConfigReq(SMnode *pMnode, SDbObj *pDb, SVgObj *pV
   alterReq.walLevel = pDb->cfg.walLevel;
   alterReq.strict = pDb->cfg.strict;
   alterReq.cacheLast = pDb->cfg.cacheLast;
+  alterReq.sttTrigger = pDb->cfg.sstTrigger;
+  alterReq.minRows = pDb->cfg.minRows;
 
   mInfo("vgId:%d, build alter vnode config req", pVgroup->vgId);
   int32_t contLen = tSerializeSAlterVnodeConfigReq(NULL, 0, &alterReq);

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -1541,6 +1541,14 @@ static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void
     }
   }
 
+  if (req.sttTrigger != -1 && req.sttTrigger != pVnode->config.sttTrigger) {
+    pVnode->config.sttTrigger = req.sttTrigger;
+  }
+
+  if (req.minRows != -1 && req.minRows != pVnode->config.tsdbCfg.minRows) {
+    pVnode->config.tsdbCfg.minRows = req.minRows;
+  }
+
   if (walChanged) {
     walAlter(pVnode->pWal, &pVnode->config.walCfg);
   }

source/libs/parser/inc/sql.y

@@ -236,6 +236,7 @@ alter_db_option(A) ::= REPLICA NK_INTEGER(B).
 //alter_db_option(A) ::= STRICT NK_STRING(B).                                       { A.type = DB_OPTION_STRICT; A.val = B; }
 alter_db_option(A) ::= WAL_LEVEL NK_INTEGER(B).                                   { A.type = DB_OPTION_WAL; A.val = B; }
 alter_db_option(A) ::= STT_TRIGGER NK_INTEGER(B).                                 { A.type = DB_OPTION_STT_TRIGGER; A.val = B; }
+alter_db_option(A) ::= MINROWS NK_INTEGER(B).                                     { A.type = DB_OPTION_MINROWS; A.val = B; }
 
 %type integer_list                                                                { SNodeList* }
 %destructor integer_list                                                          { nodesDestroyList($$); }

source/libs/parser/src/parTranslater.c

@@ -2500,9 +2500,8 @@ static int32_t translateTable(STranslateContext* pCxt, SNode* pTable) {
       STempTableNode* pTempTable = (STempTableNode*)pTable;
       code = translateSubquery(pCxt, pTempTable->pSubquery);
       if (TSDB_CODE_SUCCESS == code) {
-        if (QUERY_NODE_SELECT_STMT == nodeType(pTempTable->pSubquery) && 
-           ((SSelectStmt*)pTempTable->pSubquery)->isEmptyResult &&
-           isSelectStmt(pCxt->pCurrStmt)) {
+        if (QUERY_NODE_SELECT_STMT == nodeType(pTempTable->pSubquery) &&
+            ((SSelectStmt*)pTempTable->pSubquery)->isEmptyResult && isSelectStmt(pCxt->pCurrStmt)) {
           ((SSelectStmt*)pCxt->pCurrStmt)->isEmptyResult = true;
         }
 
@@ -4254,6 +4253,7 @@ static void buildAlterDbReq(STranslateContext* pCxt, SAlterDatabaseStmt* pStmt,
   pReq->cacheLastSize = pStmt->pOptions->cacheLastSize;
   pReq->replications = pStmt->pOptions->replica;
   pReq->sstTrigger = pStmt->pOptions->sstTrigger;
+  pReq->minRows = pStmt->pOptions->minRowsPerBlock;
   return;
 }
 

source/libs/parser/test/parAlterToBalanceTest.cpp

@@ -108,6 +108,7 @@ TEST_F(ParserInitialATest, alterDnode) {
  *   | REPLICA int_value                                         -- todo: enum 1, 3, default 1, unit replica
  *   | WAL_LEVEL int_value                                       -- enum 1, 2, default 1
  *   | STT_TRIGGER int_value                                     -- rang [1, 16], default 8
+ *   | MINROWS int_value                                         -- rang [10, 1000], default 100
  * }
  */
 TEST_F(ParserInitialATest, alterDatabase) {
@@ -133,6 +134,7 @@ TEST_F(ParserInitialATest, alterDatabase) {
     expect.cacheLastSize = -1;
     expect.replications = -1;
     expect.sstTrigger = -1;
+    expect.minRows = -1;
   };
   auto setAlterDbBuffer = [&](int32_t buffer) { expect.buffer = buffer; };
   auto setAlterDbPageSize = [&](int32_t pageSize) { expect.pageSize = pageSize; };
@@ -150,6 +152,7 @@ TEST_F(ParserInitialATest, alterDatabase) {
   auto setAlterDbCacheModel = [&](int8_t cacheModel) { expect.cacheLast = cacheModel; };
   auto setAlterDbReplica = [&](int8_t replications) { expect.replications = replications; };
   auto setAlterDbSttTrigger = [&](int8_t sstTrigger) { expect.sstTrigger = sstTrigger; };
+  auto setAlterDbMinRows = [&](int32_t minRows) { expect.minRows = minRows; };
 
   setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
     ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_ALTER_DATABASE_STMT);
@@ -170,6 +173,7 @@ TEST_F(ParserInitialATest, alterDatabase) {
     ASSERT_EQ(req.cacheLast, expect.cacheLast);
     ASSERT_EQ(req.replications, expect.replications);
     ASSERT_EQ(req.sstTrigger, expect.sstTrigger);
+    ASSERT_EQ(req.minRows, expect.minRows);
   });
 
   const int32_t MINUTE_PER_DAY = MILLISECOND_PER_DAY / MILLISECOND_PER_MINUTE;
@@ -277,6 +281,15 @@ TEST_F(ParserInitialATest, alterDatabase) {
   setAlterDbSttTrigger(16);
   run("ALTER DATABASE test STT_TRIGGER 16");
   clearAlterDbReq();
+
+  initAlterDb("test");
+  setAlterDbMinRows(10);
+  run("ALTER DATABASE test MINROWS 10");
+  setAlterDbMinRows(50);
+  run("ALTER DATABASE test MINROWS 50");
+  setAlterDbMinRows(1000);
+  run("ALTER DATABASE test MINROWS 1000");
+  clearAlterDbReq();
 }
 
 TEST_F(ParserInitialATest, alterDatabaseSemanticCheck) {

tests/script/tsim/db/alter_option.sim

@@ -285,8 +285,6 @@ sql_error alter database db keep -1
 
 print ============== modify minrows
 sql_error alter database db minrows 8
-sql_error alter database db minrows 200
-sql_error alter database db minrows 11
 sql_error alter database db minrows 8000
 sql_error alter database db minrows 8001