add mndtopic

include/common/taosmsg.h

@@ -83,6 +83,10 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SHOW, "show" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SHOW_RETRIEVE, "retrieve" )  
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_TOPIC, "create-topic" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_TOPIC, "alter-topic" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_TOPIC, "drop-topic" )
+
 // message from client to qnode
 // message from client to dnode
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_NETWORK_TEST, "nettest" )
@@ -96,6 +100,11 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_NETWORK_TEST, "nettest" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STB_IN, "create-stb-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB_IN, "alter-stb-internal" )
 TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB_IN, "drop-stb-internal" )
+
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_TOPIC_IN, "create-topic-internal" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_TOPIC_IN, "alter-topic-internal" )
+TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_TOPIC_IN, "drop-topic-internal" )
+
 // message from mnode to mnode
 // message from mnode to qnode
 // message from mnode to dnode
@@ -1140,6 +1149,40 @@ typedef struct SSchedulerCancelMsg {
   uint64_t  taskId;
 } SSchedulerCancelMsg;
 
+typedef struct {
+  char    name[TSDB_TOPIC_FNAME_LEN];
+  int8_t  igExists;
+  void*   executor;
+  char*   sql;
+} SCreateTopicMsg;
+
+typedef struct {
+  char   name[TSDB_TABLE_FNAME_LEN];
+  int8_t igNotExists;
+} SDropTopicMsg;
+
+typedef struct {
+  char    name[TSDB_TABLE_FNAME_LEN];
+  int8_t  alterType;
+  SSchema schema;
+} SAlterTopicMsg;
+
+typedef struct {
+  SMsgHead head;
+  char     name[TSDB_TABLE_FNAME_LEN];
+  uint64_t tuid;
+  int32_t  sverson;
+  int32_t  execLen;
+  char*    executor;
+  int32_t  sqlLen;
+  char*    sql;
+} SCreateTopicInternalMsg;
+
+typedef struct {
+  SMsgHead head;
+  char     name[TSDB_TABLE_FNAME_LEN];
+  uint64_t tuid;
+} SDropTopicInternalMsg;
 
 #pragma pack(pop)
 

include/common/tname.h

@@ -16,18 +16,22 @@
 #ifndef TDENGINE_TNAME_H
 #define TDENGINE_TNAME_H
 
+#include "tdef.h"
+
 #define TSDB_DB_NAME_T     1
 #define TSDB_TABLE_NAME_T  2
 
 #define T_NAME_ACCT        0x1u
 #define T_NAME_DB          0x2u
 #define T_NAME_TABLE       0x4u
+#define T_NAME_TOPIC       0x8u
 
 typedef struct SName {
   uint8_t type;  //db_name_t, table_name_t
   int32_t acctId;
   char    dbname[TSDB_DB_NAME_LEN];
   char    tname[TSDB_TABLE_NAME_LEN];
+  char    topicName[TSDB_TOPIC_NAME_LEN];
 } SName;
 
 int32_t tNameExtractFullName(const SName* name, char* dst);

include/dnode/mnode/sdb/sdb.h

@@ -16,6 +16,8 @@
 #ifndef _TD_SDB_H_
 #define _TD_SDB_H_
 
+#include "os.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -159,11 +161,12 @@ typedef enum {
   SDB_USER = 5,
   SDB_AUTH = 6,
   SDB_ACCT = 7,
-  SDB_VGROUP = 8,
-  SDB_STB = 9,
-  SDB_DB = 10,
-  SDB_FUNC = 11,
-  SDB_MAX = 12
+  SDB_TOPIC = 8,
+  SDB_VGROUP = 9,
+  SDB_STB = 10,
+  SDB_DB = 11,
+  SDB_FUNC = 12,
+  SDB_MAX = 13
 } ESdbType;
 
 typedef struct SSdb SSdb;

include/util/taoserror.h

@@ -220,6 +220,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_COLUMN_ALREAY_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03AA)
 #define TSDB_CODE_MND_COLUMN_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03AB)
 #define TSDB_CODE_MND_EXCEED_MAX_ROW_BYTES      TAOS_DEF_ERROR_CODE(0, 0x03AC)
+#define TSDB_CODE_MND_NAME_CONFLICT_WITH_TOPIC  TAOS_DEF_ERROR_CODE(0, 0x03AD)
 
 // mnode-func
 #define TSDB_CODE_MND_FUNC_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x03C0)
@@ -234,6 +235,15 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_TRANS_ALREADY_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03D0)
 #define TSDB_CODE_MND_TRANS_NOT_EXIST           TAOS_DEF_ERROR_CODE(0, 0x03D1)
 
+// mnode-topic
+#define TSDB_CODE_MND_TOPIC_ALREADY_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03E0)
+#define TSDB_CODE_MND_TOPIC_NOT_EXIST           TAOS_DEF_ERROR_CODE(0, 0x03E1)
+#define TSDB_CODE_MND_TOO_MANY_TOPICS           TAOS_DEF_ERROR_CODE(0, 0x03E2)
+#define TSDB_CODE_MND_INVALID_TOPIC             TAOS_DEF_ERROR_CODE(0, 0x03E3)
+#define TSDB_CODE_MND_INVALID_TOPIC_OPTION      TAOS_DEF_ERROR_CODE(0, 0x03E4)
+#define TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED    TAOS_DEF_ERROR_CODE(0, 0x03E5)
+#define TSDB_CODE_MND_NAME_CONFLICT_WITH_STB    TAOS_DEF_ERROR_CODE(0, 0x03E6)
+
 // dnode
 #define TSDB_CODE_DND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0400)
 #define TSDB_CODE_DND_EXITING                   TAOS_DEF_ERROR_CODE(0, 0x0401)

include/util/tdef.h

@@ -163,6 +163,7 @@ do { \
 
 #define TSDB_NODE_NAME_LEN        64
 #define TSDB_TABLE_NAME_LEN       193     // it is a null-terminated string
+#define TSDB_TOPIC_NAME_LEN       193     // it is a null-terminated string
 #define TSDB_DB_NAME_LEN          65
 #define TSDB_FULL_DB_NAME_LEN     (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN)
 
@@ -175,6 +176,7 @@ do { \
 
 #define TSDB_TYPE_STR_MAX_LEN     32
 #define TSDB_TABLE_FNAME_LEN      (TSDB_FULL_DB_NAME_LEN + TSDB_TABLE_NAME_LEN)
+#define TSDB_TOPIC_FNAME_LEN      TSDB_TABLE_FNAME_LEN
 #define TSDB_COL_NAME_LEN         65
 #define TSDB_MAX_SAVED_SQL_LEN    TSDB_MAX_COLUMNS * 64
 #define TSDB_MAX_SQL_LEN          TSDB_PAYLOAD_SIZE

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

@@ -285,6 +285,21 @@ typedef struct {
   char    payload[];
 } SShowObj;
 
+typedef struct {
+  char name[TSDB_TOPIC_FNAME_LEN];
+  char db[TSDB_FULL_DB_NAME_LEN];
+  int64_t createTime;
+  int64_t updateTime;
+  uint64_t uid;
+  uint64_t dbUid;
+  int32_t version;
+  SRWLatch lock;
+  int32_t  execLen;
+  void*    executor;
+  int32_t  sqlLen;
+  char*    sql;
+} STopicObj;
+
 typedef struct SMnodeMsg {
   char    user[TSDB_USER_LEN];
   char    db[TSDB_FULL_DB_NAME_LEN];

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

@@ -25,6 +25,9 @@ extern "C" {
 int32_t mndInitStb(SMnode *pMnode);
 void    mndCleanupStb(SMnode *pMnode);
 
+SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName);
+void mndReleaseStb(SMnode *pMnode, SStbObj *pStb);
+
 #ifdef __cplusplus
 }
 #endif

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_MND_STB_H_
+#define _TD_MND_STB_H_
+
+#include "mndInt.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int32_t mndInitTopic(SMnode *pMnode);
+void    mndCleanupTopic(SMnode *pMnode);
+
+SStbObj *mndAcquireTopic(SMnode *pMnode, char *topicName);
+void mndReleaseTopic(SMnode *pMnode, STopicObj *pTopic);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_MND_STB_H_*/

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

@@ -484,6 +484,16 @@ static int32_t mndProcessCreateStbMsg(SMnodeMsg *pMsg) {
     }
   }
 
+  //topic should have different name with stb
+  SStbObj *pTopic = mndAcquireStb(pMnode, pCreate->name);
+  if (pTopic != NULL) {
+    sdbRelease(pMnode->pSdb, pTopic);
+    terrno = TSDB_CODE_MND_NAME_CONFLICT_WITH_TOPIC;
+    mError("stb:%s, failed to create since %s", pCreate->name, terrstr());
+    return -1;
+  }
+  sdbRelease(pMnode->pSdb, pTopic);
+
   SDbObj *pDb = mndAcquireDbByStb(pMnode, pCreate->name);
   if (pDb == NULL) {
     terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
@@ -891,4 +901,4 @@ static int32_t mndRetrieveStb(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int3
 static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
   SSdb *pSdb = pMnode->pSdb;
   sdbCancelFetch(pSdb, pIter);
-}
\ No newline at end of file
+}