Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/vnode_refact1

include/common/tmsg.h

@@ -1281,8 +1281,9 @@ typedef struct {
 #define STREAM_TRIGGER_WINDOW_CLOSE 2
 
 typedef struct {
-  char    name[TSDB_TOPIC_FNAME_LEN];
-  char    outputSTbName[TSDB_TABLE_FNAME_LEN];
+  char    name[TSDB_TABLE_FNAME_LEN];
+  char    sourceDB[TSDB_DB_FNAME_LEN];
+  char    targetStbFullName[TSDB_TABLE_FNAME_LEN];
   int8_t  igExists;
   char*   sql;
   char*   ast;

include/dnode/mnode/sdb/sdb.h

@@ -331,7 +331,6 @@ int32_t  sdbGetRawSoftVer(SSdbRaw *pRaw, int8_t *sver);
 int32_t  sdbGetRawTotalSize(SSdbRaw *pRaw);
 
 SSdbRow *sdbAllocRow(int32_t objSize);
-void     sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc);
 void    *sdbGetRowObj(SSdbRow *pRow);
 
 typedef struct SSdb {

source/client/src/tmq.c

@@ -739,7 +739,7 @@ TAOS_RES* tmq_create_stream(TAOS* taos, const char* streamName, const char* tbNa
       .sql = (char*)sql,
   };
   tNameExtractFullName(&name, req.name);
-  strcpy(req.outputSTbName, tbName);
+  strcpy(req.targetStbFullName, tbName);
 
   int   tlen = tSerializeSCMCreateStreamReq(NULL, 0, &req);
   void* buf = taosMemoryMalloc(tlen);

source/common/src/tmsg.c

@@ -3554,10 +3554,12 @@ int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateS
 
   if (tStartEncode(&encoder) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
-  if (tEncodeCStr(&encoder, pReq->outputSTbName) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->targetStbFullName) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->igExists) < 0) return -1;
   if (tEncodeI32(&encoder, sqlLen) < 0) return -1;
   if (tEncodeI32(&encoder, astLen) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->triggerType) < 0) return -1;
+  if (tEncodeI64(&encoder, pReq->watermark) < 0) return -1;
   if (sqlLen > 0 && tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
   if (astLen > 0 && tEncodeCStr(&encoder, pReq->ast) < 0) return -1;
 
@@ -3577,7 +3579,7 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
 
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
-  if (tDecodeCStrTo(&decoder, pReq->outputSTbName) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->targetStbFullName) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->igExists) < 0) return -1;
   if (tDecodeI32(&decoder, &sqlLen) < 0) return -1;
   if (tDecodeI32(&decoder, &astLen) < 0) return -1;
@@ -3905,4 +3907,4 @@ int32_t tDecodeSVDropStbReq(SCoder *pCoder, SVDropStbReq *pReq) {
 
   tEndDecode(pCoder);
   return 0;
-}
\ No newline at end of file
+}

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

@@ -295,7 +295,7 @@ static int32_t mndCreateStream(SMnode *pMnode, SNodeMsg *pReq, SCMCreateStreamRe
   SStreamObj streamObj = {0};
   tstrncpy(streamObj.name, pCreate->name, TSDB_STREAM_FNAME_LEN);
   tstrncpy(streamObj.sourceDb, pDb->name, TSDB_DB_FNAME_LEN);
-  tstrncpy(streamObj.targetSTbName, pCreate->outputSTbName, TSDB_TABLE_FNAME_LEN);
+  tstrncpy(streamObj.targetSTbName, pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN);
   streamObj.createTime = taosGetTimestampMs();
   streamObj.updateTime = streamObj.createTime;
   streamObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));

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

@@ -39,14 +39,16 @@ static int32_t  mndRetrieveUsers(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *p
 static void     mndCancelGetNextUser(SMnode *pMnode, void *pIter);
 
 int32_t mndInitUser(SMnode *pMnode) {
-  SSdbTable table = {.sdbType = SDB_USER,
-                     .keyType = SDB_KEY_BINARY,
-                     .deployFp = (SdbDeployFp)mndCreateDefaultUsers,
-                     .encodeFp = (SdbEncodeFp)mndUserActionEncode,
-                     .decodeFp = (SdbDecodeFp)mndUserActionDecode,
-                     .insertFp = (SdbInsertFp)mndUserActionInsert,
-                     .updateFp = (SdbUpdateFp)mndUserActionUpdate,
-                     .deleteFp = (SdbDeleteFp)mndUserActionDelete};
+  SSdbTable table = {
+      .sdbType = SDB_USER,
+      .keyType = SDB_KEY_BINARY,
+      .deployFp = (SdbDeployFp)mndCreateDefaultUsers,
+      .encodeFp = (SdbEncodeFp)mndUserActionEncode,
+      .decodeFp = (SdbDecodeFp)mndUserActionDecode,
+      .insertFp = (SdbInsertFp)mndUserActionInsert,
+      .updateFp = (SdbUpdateFp)mndUserActionUpdate,
+      .deleteFp = (SdbDeleteFp)mndUserActionDelete,
+  };
 
   mndSetMsgHandle(pMnode, TDMT_MND_CREATE_USER, mndProcessCreateUserReq);
   mndSetMsgHandle(pMnode, TDMT_MND_ALTER_USER, mndProcessAlterUserReq);

source/dnode/mnode/impl/test/sdb/sdbTest.cpp

@@ -9,75 +9,314 @@
  *
  */
 
-#include "sut.h"
+#include <gtest/gtest.h>
 
-class MndTestShow : public ::testing::Test {
- protected:
-  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_show", 9021); }
-  static void TearDownTestSuite() { test.Cleanup(); }
+#include "sdb.h"
 
-  static Testbase test;
+class MndTestSdb : public ::testing::Test {
+ protected:
+  static void SetUpTestSuite() {}
+  static void TearDownTestSuite() {}
 
  public:
   void SetUp() override {}
   void TearDown() override {}
 };
 
-Testbase MndTestShow::test;
+typedef struct SMnode {
+  int32_t v100;
+  int32_t v200;
+  SSdb   *pSdb;
+} SMnode;
+
+typedef struct SStrObj {
+  char    key[24];
+  int8_t  v8;
+  int16_t v16;
+  int32_t v32;
+  int64_t v64;
+  char    vstr[32];
+  char    unused[48];
+} SStrObj;
 
-TEST_F(MndTestShow, 01_ShowMsg_InvalidMsgMax) {
-  SShowReq showReq = {0};
-  showReq.type = TSDB_MGMT_TABLE_MAX;
+typedef struct SI32Obj {
+  int32_t key;
+  int8_t  v8;
+  int16_t v16;
+  int32_t v32;
+  int64_t v64;
+  char    vstr[32];
+  char    unused[48];
+} SI32Obj;
 
-  int32_t contLen = tSerializeSShowReq(NULL, 0, &showReq);
-  void*   pReq = rpcMallocCont(contLen);
-  tSerializeSShowReq(pReq, contLen, &showReq);
-  tFreeSShowReq(&showReq);
+typedef struct SI64Obj {
+  int64_t key;
+  int8_t  v8;
+  int16_t v16;
+  int32_t v32;
+  int64_t v64;
+  char    vstr[32];
+  char    unused[48];
+} SI64Obj;
 
-  SRpcMsg* pRsp = test.SendReq(TDMT_MND_SYSTABLE_RETRIEVE, pReq, contLen);
-  ASSERT_NE(pRsp, nullptr);
-  ASSERT_NE(pRsp->code, 0);
+SSdbRaw *strEncode(SStrObj *pObj) {
+  int32_t  dataPos = 0;
+  SSdbRaw *pRaw = sdbAllocRaw(SDB_USER, 1, sizeof(SStrObj));
+
+  sdbSetRawBinary(pRaw, dataPos, pObj->key, sizeof(pObj->key));
+  dataPos += sizeof(pObj->key);
+  sdbSetRawInt8(pRaw, dataPos, pObj->v8);
+  dataPos += sizeof(pObj->v8);
+  sdbSetRawInt16(pRaw, dataPos, pObj->v16);
+  dataPos += sizeof(pObj->v16);
+  sdbSetRawInt32(pRaw, dataPos, pObj->v32);
+  dataPos += sizeof(pObj->v32);
+  sdbSetRawInt64(pRaw, dataPos, pObj->v64);
+  dataPos += sizeof(pObj->v64);
+  sdbSetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr));
+  dataPos += sizeof(pObj->vstr);
+  sdbSetRawDataLen(pRaw, dataPos);
+
+  return pRaw;
 }
 
-TEST_F(MndTestShow, 02_ShowMsg_InvalidMsgStart) {
-  SShowReq showReq = {0};
-  showReq.type = TSDB_MGMT_TABLE_START;
+SSdbRow *strDecode(SSdbRaw *pRaw) {
+  int8_t sver = 0;
+  if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
+  if (sver != 1) return NULL;
+
+  SSdbRow *pRow = sdbAllocRow(sizeof(SStrObj));
+  if (pRow == NULL) return NULL;
 
-  int32_t contLen = tSerializeSShowReq(NULL, 0, &showReq);
-  void*   pReq = rpcMallocCont(contLen);
-  tSerializeSShowReq(pReq, contLen, &showReq);
-  tFreeSShowReq(&showReq);
+  SStrObj *pObj = (SStrObj *)sdbGetRowObj(pRow);
+  if (pObj == NULL) return NULL;
 
-  SRpcMsg* pRsp = test.SendReq(TDMT_MND_SYSTABLE_RETRIEVE, pReq, contLen);
-  ASSERT_NE(pRsp, nullptr);
-  ASSERT_NE(pRsp->code, 0);
+  int32_t dataPos = 0;
+  sdbGetRawBinary(pRaw, dataPos, pObj->key, sizeof(pObj->key));
+  dataPos += sizeof(pObj->key);
+  sdbGetRawInt8(pRaw, dataPos, &pObj->v8);
+  dataPos += sizeof(pObj->v8);
+  sdbGetRawInt16(pRaw, dataPos, &pObj->v16);
+  dataPos += sizeof(pObj->v16);
+  sdbGetRawInt32(pRaw, dataPos, &pObj->v32);
+  dataPos += sizeof(pObj->v32);
+  sdbGetRawInt64(pRaw, dataPos, &pObj->v64);
+  dataPos += sizeof(pObj->v64);
+  sdbGetRawBinary(pRaw, dataPos, pObj->vstr, sizeof(pObj->vstr));
+  dataPos += sizeof(pObj->vstr);
+
+  return pRow;
 }
 
-TEST_F(MndTestShow, 03_ShowMsg_Conn) {
-  char passwd[] = "taosdata";
-  char secretEncrypt[TSDB_PASSWORD_LEN] = {0};
-  taosEncryptPass_c((uint8_t*)passwd, strlen(passwd), secretEncrypt);
+int32_t strInsert(SSdb *pSdb, SStrObj *pObj) { return 0; }
+
+int32_t strDelete(SSdb *pSdb, SStrObj *pObj, bool callFunc) { return 0; }
 
-  SConnectReq connectReq = {0};
-  connectReq.pid = 1234;
-  strcpy(connectReq.app, "mnode_test_show");
-  strcpy(connectReq.db, "");
-  strcpy(connectReq.user, "root");
-  strcpy(connectReq.passwd, secretEncrypt);
+int32_t strUpdate(SSdb *pSdb, SStrObj *pOld, SStrObj *pNew) {
+  pOld->v8 = pNew->v8;
+  pOld->v16 = pNew->v16;
+  pOld->v32 = pNew->v32;
+  pOld->v64 = pNew->v64;
+  strcpy(pOld->vstr, pNew->vstr);
+  return 0;
+}
+
+void strSetDefault(SStrObj *pObj, int32_t index) {
+  memset(pObj, 0, sizeof(SStrObj));
+  snprintf(pObj->key, sizeof(pObj->key), "k%d", index * 1000);
+  pObj->v8 = index;
+  pObj->v16 = index;
+  pObj->v32 = index * 1000;
+  pObj->v64 = index * 1000;
+  snprintf(pObj->vstr, sizeof(pObj->vstr), "v%d", index * 1000);
+}
 
-  int32_t contLen = tSerializeSConnectReq(NULL, 0, &connectReq);
-  void*   pReq = rpcMallocCont(contLen);
-  tSerializeSConnectReq(pReq, contLen, &connectReq);
+int32_t strDefault(SMnode *pMnode) {
+  SStrObj  strObj;
+  SSdbRaw *pRaw = NULL;
 
-  SRpcMsg* pRsp = test.SendReq(TDMT_MND_CONNECT, pReq, contLen);
-  ASSERT_NE(pRsp, nullptr);
-  ASSERT_EQ(pRsp->code, 0);
+  strSetDefault(&strObj, 1);
+  pRaw = strEncode(&strObj);
+  sdbSetRawStatus(pRaw, SDB_STATUS_READY);
+  if (sdbWrite(pMnode->pSdb, pRaw) != 0) return -1;
 
-  test.SendShowReq(TSDB_MGMT_TABLE_CONNS, "connections", "");
-  // EXPECT_EQ(test.GetShowRows(), 1);
+  strSetDefault(&strObj, 2);
+  pRaw = strEncode(&strObj);
+  sdbSetRawStatus(pRaw, SDB_STATUS_READY);
+  if (sdbWriteWithoutFree(pMnode->pSdb, pRaw) != 0) return -1;
+  sdbFreeRaw(pRaw);
+
+  return 0;
+}
+
+bool sdbTraverseSucc1(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) {
+  if (pObj->v8 == 1) {
+    *p1 = *p2 + *p3 + pObj->v8;
+  }
+  return true;
+}
+
+bool sdbTraverseSucc2(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) {
+  *p1 = *p2 + *p3 + pObj->v8;
+  return true;
 }
 
-TEST_F(MndTestShow, 04_ShowMsg_Cluster) {
-  test.SendShowReq(TSDB_MGMT_TABLE_CLUSTER, "cluster", "");
-  EXPECT_EQ(test.GetShowRows(), 1);
+bool sdbTraverseFail(SMnode *pMnode, SStrObj *pObj, int32_t *p1, int32_t *p2, int32_t *p3) {
+  *p1 = *p2 + *p3;
+  return false;
 }
+
+TEST_F(MndTestSdb, 01_Write) {
+  void    *pIter;
+  int32_t  num;
+  SStrObj *pObj;
+  SMnode   mnode;
+  SSdb    *pSdb;
+  SSdbOpt  opt = {0};
+  int32_t  p1 = 0;
+  int32_t  p2 = 111;
+  int32_t  p3 = 222;
+
+  mnode.v100 = 100;
+  mnode.v200 = 200;
+  opt.pMnode = &mnode;
+  opt.path = "/tmp/mnode_test_sdb";
+  taosRemoveDir(opt.path);
+
+  SSdbTable strTable = {
+      .sdbType = SDB_USER,
+      .keyType = SDB_KEY_BINARY,
+      .deployFp = (SdbDeployFp)strDefault,
+      .encodeFp = (SdbEncodeFp)strEncode,
+      .decodeFp = (SdbDecodeFp)strDecode,
+      .insertFp = (SdbInsertFp)strInsert,
+      .updateFp = (SdbUpdateFp)strUpdate,
+      .deleteFp = (SdbDeleteFp)strDelete,
+  };
+
+  pSdb = sdbInit(&opt);
+  mnode.pSdb = pSdb;
+
+  ASSERT_NE(pSdb, nullptr);
+  ASSERT_EQ(sdbSetTable(pSdb, strTable), 0);
+  ASSERT_EQ(sdbDeploy(pSdb), 0);
+#if 0
+  pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k1000");
+  ASSERT_NE(pObj, nullptr);
+  EXPECT_STREQ(pObj->key, "k1000");
+  EXPECT_STREQ(pObj->vstr, "v1000");
+  EXPECT_EQ(pObj->v8, 1);
+  EXPECT_EQ(pObj->v16, 1);
+  EXPECT_EQ(pObj->v32, 1000);
+  EXPECT_EQ(pObj->v64, 1000);
+  sdbRelease(pSdb, pObj);
+
+  pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k2000");
+  ASSERT_NE(pObj, nullptr);
+  EXPECT_STREQ(pObj->key, "k2000");
+  EXPECT_STREQ(pObj->vstr, "v2000");
+  EXPECT_EQ(pObj->v8, 2);
+  EXPECT_EQ(pObj->v16, 2);
+  EXPECT_EQ(pObj->v32, 2000);
+  EXPECT_EQ(pObj->v64, 2000);
+  sdbRelease(pSdb, pObj);
+
+  pObj = (SStrObj *)sdbAcquire(pSdb, SDB_USER, "k200");
+  ASSERT_EQ(pObj, nullptr);
+
+  pIter = NULL;
+  num = 0;
+  do {
+    pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj);
+    if (pIter == NULL) break;
+    ASSERT_NE(pObj, nullptr);
+    num++;
+    sdbRelease(pSdb, pObj);
+  } while (1);
+  EXPECT_EQ(num, 2);
+
+  do {
+    pIter = sdbFetch(pSdb, SDB_USER, pIter, (void **)&pObj);
+    if (pIter == NULL) break;
+    if (strcmp(pObj->key, "k1000") == 0) {
+      sdbCancelFetch(pSdb, pIter);
+      break;
+    }
+  } while (1);
+  EXPECT_STREQ(pObj->key, "k1000");
+
+  p1 = 0;
+  p2 = 111;
+  p3 = 222;
+  sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3);
+  EXPECT_EQ(p1, 334);
+
+  p1 = 0;
+  p2 = 111;
+  p3 = 222;
+  sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseSucc2, &p1, &p2, &p3);
+  EXPECT_EQ(p1, 669);
+
+  p1 = 0;
+  p2 = 111;
+  p3 = 222;
+  sdbTraverse(pSdb, SDB_USER, (sdbTraverseFp)sdbTraverseFail, &p1, &p2, &p3);
+  EXPECT_EQ(p1, 333);
+
+  EXPECT_EQ(sdbGetSize(pSdb, SDB_USER), 2);
+  EXPECT_EQ(sdbGetMaxId(pSdb, SDB_USER), -1);
+  EXPECT_EQ(sdbGetTableVer(pSdb, SDB_USER), 2);
+  EXPECT_EQ(sdbUpdateVer(pSdb, 0), 2);
+  EXPECT_EQ(sdbUpdateVer(pSdb, 1), 3);
+  EXPECT_EQ(sdbUpdateVer(pSdb, -1), 2);
+
+  // insert, call func
+
+  // update, call func
+
+  // delete, call func 2
+
+  // write version
+
+  // sdb Write ver
+
+  // sdbRead
+#endif
+  ASSERT_EQ(sdbWriteFile(pSdb), 0);
+  sdbCleanup(pSdb);
+}
+
+TEST_F(MndTestSdb, 01_Read) {
+  void    *pIter;
+  int32_t  num;
+  SStrObj *pObj;
+  SMnode   mnode;
+  SSdb    *pSdb;
+  SSdbOpt  opt = {0};
+  int32_t  p1 = 0;
+  int32_t  p2 = 111;
+  int32_t  p3 = 222;
+
+  mnode.v100 = 100;
+  mnode.v200 = 200;
+  opt.pMnode = &mnode;
+  opt.path = "/tmp/mnode_test_sdb";
+  taosRemoveDir(opt.path);
+
+  SSdbTable strTable = {
+      .sdbType = SDB_USER,
+      .keyType = SDB_KEY_BINARY,
+      .deployFp = (SdbDeployFp)strDefault,
+      .encodeFp = (SdbEncodeFp)strEncode,
+      .decodeFp = (SdbDecodeFp)strDecode,
+      .insertFp = (SdbInsertFp)strInsert,
+      .updateFp = (SdbUpdateFp)strDelete,
+      .deleteFp = (SdbDeleteFp)strUpdate,
+  };
+
+  pSdb = sdbInit(&opt);
+  mnode.pSdb = pSdb;
+
+  ASSERT_EQ(sdbReadFile(pSdb), 0);
+  sdbCleanup(pSdb);
+}
\ No newline at end of file

source/dnode/mnode/sdb/inc/sdbInt.h

@@ -52,6 +52,8 @@ typedef struct SSdbRow {
 const char *sdbTableName(ESdbType type);
 void        sdbPrintOper(SSdb *pSdb, SSdbRow *pRow, const char *oper);
 
+void sdbFreeRow(SSdb *pSdb, SSdbRow *pRow, bool callFunc);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mnode/sdb/src/sdb.c

@@ -141,7 +141,7 @@ int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {
 }
 
 static int32_t sdbCreateDir(SSdb *pSdb) {
-  if (taosMkDir(pSdb->currDir) != 0) {
+  if (taosMulMkDir(pSdb->currDir) != 0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     mError("failed to create dir:%s since %s", pSdb->currDir, terrstr());
     return -1;

source/libs/parser/src/parTranslater.c

@@ -2658,7 +2658,7 @@ static int32_t translateCreateStream(STranslateContext* pCxt, SCreateStreamStmt*
   if ('\0' != pStmt->targetTabName[0]) {
     strcpy(name.dbname, pStmt->targetDbName);
     strcpy(name.tname, pStmt->targetTabName);
-    tNameExtractFullName(&name, createReq.outputSTbName);
+    tNameExtractFullName(&name, createReq.targetStbFullName);
   }
 
   int32_t code = translateQuery(pCxt, pStmt->pQuery);

source/os/src/osDir.c

@@ -72,8 +72,8 @@ void taosRemoveDir(const char *dirname) {
   while ((de = taosReadDir(pDir)) != NULL) {
     if (strcmp(taosGetDirEntryName(de), ".") == 0 || strcmp(taosGetDirEntryName(de), "..") == 0) continue;
 
-    char filename[1024];
-    snprintf(filename, sizeof(filename), "%s/%s", dirname, taosGetDirEntryName(de));
+    char filename[1024] = {0};
+    snprintf(filename, sizeof(filename), "%s%s%s", dirname, TD_DIRSEP, taosGetDirEntryName(de));
     if (taosDirEntryIsDir(de)) {
       taosRemoveDir(filename);
     } else {